[td-11818] refactor, fix bug in select

source/dnode/vnode/inc/tsdb.h

@@ -75,7 +75,6 @@ typedef struct STsdbQueryCond {
 } STsdbQueryCond;
 
 typedef struct {
-  void    *pTable;
   TSKEY    lastKey;
   uint64_t uid;
 } STableKeyInfo;

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -81,7 +81,6 @@ enum {
   CHECKINFO_CHOSEN_BOTH = 2    //for update=2(merge case)
 };
 
-
 typedef struct STableCheckInfo {
   uint64_t      tableId;
   TSKEY         lastKey;
@@ -258,9 +257,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S
     for (int32_t j = 0; j < gsize; ++j) {
       STableKeyInfo* pKeyInfo = (STableKeyInfo*) taosArrayGet(group, j);
 
-      STableCheckInfo info = { .lastKey = pKeyInfo->lastKey};
-
-      info.tableId = pKeyInfo->uid;
+      STableCheckInfo info = { .lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid};
       if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
         if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) {
           info.lastKey = pTsdbReadHandle->window.skey;
@@ -277,12 +274,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S
   }
 
   // TODO  group table according to the tag value.
-//  taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar);
-//  size_t gsize = taosArrayGetSize(pTableCheckInfo);
-//  for (int32_t i = 0; i < gsize; ++i) {
-//    STableCheckInfo* pInfo = (STableCheckInfo*) taosArrayGet(pTableCheckInfo, i);
-//  }
-
+  taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar);
   return pTableCheckInfo;
 }
 
@@ -2605,7 +2597,7 @@ static int32_t getAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) {
       break;
     }
 
-    STableKeyInfo info = {.pTable = NULL, .lastKey = TSKEY_INITIAL_VAL, uid = id};
+    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
     taosArrayPush(list, &info);
   }
 
@@ -3196,7 +3188,7 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) {
       if (key < lastKey) {
         key = lastKey;
 
-        keyInfo.pTable  = pInfo->pTable;
+//        keyInfo.pTable  = pInfo->pTable;
         keyInfo.lastKey = key;
         pInfo->lastKey  = key;
 
@@ -3210,29 +3202,19 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) {
       }
     }
 
-    // clear current group, unref unused table
-    for (int32_t i = 0; i < numOfTables; ++i) {
-      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i);
-
-      // keyInfo.pTable may be NULL here.
-      if (pInfo->pTable != keyInfo.pTable) {
-//        tsdbUnRefTable(pInfo->pTable);
-      }
-    }
-
     // more than one table in each group, only one table left for each group
-    if (keyInfo.pTable != NULL) {
-      totalNumOfTable++;
-      if (taosArrayGetSize(pGroup) == 1) {
-        // do nothing
-      } else {
-        taosArrayClear(pGroup);
-        taosArrayPush(pGroup, &keyInfo);
-      }
-    } else {  // mark all the empty groups, and remove it later
-      taosArrayDestroy(pGroup);
-      taosArrayPush(emptyGroup, &j);
-    }
+//    if (keyInfo.pTable != NULL) {
+//      totalNumOfTable++;
+//      if (taosArrayGetSize(pGroup) == 1) {
+//        // do nothing
+//      } else {
+//        taosArrayClear(pGroup);
+//        taosArrayPush(pGroup, &keyInfo);
+//      }
+//    } else {  // mark all the empty groups, and remove it later
+//      taosArrayDestroy(pGroup);
+//      taosArrayPush(emptyGroup, &j);
+//    }
   }
 
   // window does not being updated, so set the original
@@ -3422,11 +3404,13 @@ void filterPrepare(void* expr, void* param) {
   }
 }
 
+#endif
 
 static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) {
+#if 0
   STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param;
-  STable* pTable1 = ((STableKeyInfo*) p1)->pTable;
-  STable* pTable2 = ((STableKeyInfo*) p2)->pTable;
+  STable* pTable1 = ((STableKeyInfo*) p1)->uid;
+  STable* pTable2 = ((STableKeyInfo*) p2)->uid;
 
   for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) {
     SColIndex* pColIndex = &pTableGroupSupp->pCols[i];
@@ -3474,10 +3458,9 @@ static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *pa
       return ret;
     }
   }
-
+#endif
   return 0;
 }
-#endif
 
 static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
   if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) {
@@ -3493,10 +3476,9 @@ static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
 void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTables, TSKEY skey,
                           STableGroupSupporter* pSupp, __ext_compar_fn_t compareFn) {
   STable* pTable = taosArrayGetP(pTableList, 0);
-
   SArray* g = taosArrayInit(16, sizeof(STableKeyInfo));
 
-  STableKeyInfo info = {.pTable = pTable, .lastKey = skey};
+  STableKeyInfo info = {.lastKey = skey};
   taosArrayPush(g, &info);
 
   for (int32_t i = 1; i < numOfTables; ++i) {
@@ -3507,13 +3489,13 @@ void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTable
     assert(ret == 0 || ret == -1);
 
     if (ret == 0) {
-      STableKeyInfo info1 = {.pTable = *p, .lastKey = skey};
+      STableKeyInfo info1 = {.lastKey = skey};
       taosArrayPush(g, &info1);
     } else {
       taosArrayPush(pGroups, &g);  // current group is ended, start a new group
       g = taosArrayInit(16, sizeof(STableKeyInfo));
 
-      STableKeyInfo info1 = {.pTable = *p, .lastKey = skey};
+      STableKeyInfo info1 = {.lastKey = skey};
       taosArrayPush(g, &info1);
     }
   }
@@ -3546,8 +3528,8 @@ SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColInd
     sup.pTagSchema = pTagSchema->pSchema;
     sup.pCols = pCols;
 
-//    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
-//    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
+    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
+    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
   }
 
   return pTableGroup;

source/libs/executor/inc/executil.h

@@ -38,7 +38,7 @@
 #define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
 #define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES)
 
-#define GET_TASKID(_t)  (((SExecTaskInfo*)(_t))->id.idstr)
+#define GET_TASKID(_t)  (((SExecTaskInfo*)(_t))->id.str)
 
 #define curTimeWindowIndex(_winres)        ((_winres)->curIndex)
 

source/libs/executor/inc/executorimpl.h

@@ -239,8 +239,7 @@ typedef struct STaskIdInfo {
   uint64_t       queryId;    // this is also a request id
   uint64_t       subplanId;
   uint64_t       templateId;
-  uint64_t       taskId;     // this is a subplan id
-  char          *idstr;
+  char          *str;
 } STaskIdInfo;
 
 typedef struct SExecTaskInfo {
@@ -639,7 +638,6 @@ int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg);
 bool isTaskKilled(SExecTaskInfo *pTaskInfo);
 int32_t checkForQueryBuf(size_t numOfTables);
 bool checkNeedToCompressQueryCol(SQInfo *pQInfo);
-bool doBuildResCheck(SQInfo* pQInfo);
 void setQueryStatus(STaskRuntimeEnv *pRuntimeEnv, int8_t status);
 
 bool onlyQueryTags(STaskAttr* pQueryAttr);

source/libs/executor/src/executor.c

@@ -18,20 +18,20 @@
 #include "executorimpl.h"
 #include "planner.h"
 
-static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, uint64_t reqId) {
+static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) {
   ASSERT(pOperator != NULL);
   if (pOperator->operatorType != OP_StreamScan) {
     if (pOperator->numOfDownstream == 0) {
-      qError("failed to find stream scan operator to set the input data block, reqId:0x%" PRIx64, reqId);
+      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);
       return TSDB_CODE_QRY_APP_ERROR;
     }
 
     if (pOperator->numOfDownstream > 1) {  // not handle this in join query
-      qError("join not supported for stream block scan, reqId:0x%" PRIx64, reqId);
+      qError("join not supported for stream block scan, %s" PRIx64, id);
       return TSDB_CODE_QRY_APP_ERROR;
     }
 
-    return doSetStreamBlock(pOperator->pDownstream[0], input, reqId);
+    return doSetStreamBlock(pOperator->pDownstream[0], input, id);
   } else {
     SStreamBlockScanInfo* pInfo = pOperator->info;
     tqReadHandleSetMsg(pInfo->readerHandle, input, 0);
@@ -50,7 +50,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input) {
 
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo;
 
-  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, pTaskInfo->id.queryId);
+  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, GET_TASKID(pTaskInfo));
   if (code != TSDB_CODE_SUCCESS) {
     qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
   } else {

source/libs/executor/src/executorimpl.c

@@ -5146,7 +5146,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) {
 
   size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
   if (pExchangeInfo->current >= totalSources) {
-    qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources,
+    qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources,
            pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0);
     return NULL;
   }
@@ -5208,7 +5208,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) {
         int64_t el = taosGetTimestampUs() - startTs;
         pExchangeInfo->totalElapsed += el;
 
-        qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources,
+        qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources,
                pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0);
         return NULL;
       } else {
@@ -7741,11 +7741,10 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId) {
 
   pTaskInfo->cost.created = taosGetTimestampMs();
   pTaskInfo->id.queryId = queryId;
-  pTaskInfo->id.taskId  = taskId;
 
   char* p = calloc(1, 128);
   snprintf(p, 128, "TID:0x%"PRIx64" QID:0x%"PRIx64, taskId, queryId);
-  pTaskInfo->id.idstr = strdup(p);
+  pTaskInfo->id.str = strdup(p);
 
   return pTaskInfo;
 }
@@ -7832,7 +7831,7 @@ static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode,
 
     SArray* pa = taosArrayInit(1, sizeof(STableKeyInfo));
 
-    STableKeyInfo info = {.pTable = NULL, .lastKey = 0, .uid = uid};
+    STableKeyInfo info = {.lastKey = 0, .uid = uid};
     taosArrayPush(pa, &info);
     taosArrayPush(groupInfo.pGroupList, &pa);
   }
@@ -8827,33 +8826,15 @@ void* freeColumnInfo(SColumnInfo* pColumnInfo, int32_t numOfCols) {
 }
 
 void doDestroyTask(SExecTaskInfo *pTaskInfo) {
-  qDebug("%s start to free execTask", GET_TASKID(pTaskInfo));
   doDestroyTableQueryInfo(&pTaskInfo->tableqinfoGroupInfo);
 //  taosArrayDestroy(pTaskInfo->summary.queryProfEvents);
 //  taosHashCleanup(pTaskInfo->summary.operatorProfResults);
 
+  tfree(pTaskInfo->sql);
+  tfree(pTaskInfo->id.str);
   qDebug("%s execTask is freed", GET_TASKID(pTaskInfo));
-  tfree(pTaskInfo);
-}
-
-bool doBuildResCheck(SQInfo* pQInfo) {
-  bool buildRes = false;
-
-  pthread_mutex_lock(&pQInfo->lock);
 
-  pQInfo->dataReady = QUERY_RESULT_READY;
-  buildRes = needBuildResAfterQueryComplete(pQInfo);
-
-  // clear qhandle owner, it must be in the secure area. other thread may run ahead before current, after it is
-  // put into task to be executed.
-  assert(pQInfo->owner == taosGetSelfPthreadId());
-  pQInfo->owner = 0;
-
-  pthread_mutex_unlock(&pQInfo->lock);
-
-  // used in retrieve blocking model.
-  tsem_post(&pQInfo->ready);
-  return buildRes;
+  tfree(pTaskInfo);
 }
 
 static void doSetTagValueToResultBuf(char* output, const char* val, int16_t type, int16_t bytes) {