Merge pull request #16840 from taosdata/enh/3.0_planner_optimize

enh: the delete physical plan increases the timestamp interval of the actual deleted data

include/libs/nodes/plannodes.h

@@ -151,6 +151,8 @@ typedef struct SVnodeModifyLogicNode {
   SArray*          pDataBlocks;
   SVgDataBlocks*   pVgDataBlocks;
   SNode*           pAffectedRows;  // SColumnNode
+  SNode*           pStartTs;       // SColumnNode
+  SNode*           pEndTs;         // SColumnNode
   uint64_t         tableId;
   uint64_t         stableId;
   int8_t           tableType;  // table type
@@ -525,6 +527,8 @@ typedef struct SDataDeleterNode {
   char          tsColName[TSDB_COL_NAME_LEN];
   STimeWindow   deleteTimeRange;
   SNode*        pAffectedRows;
+  SNode*        pStartTs;
+  SNode*        pEndTs;
 } SDataDeleterNode;
 
 typedef struct SSubplan {

include/libs/nodes/querynodes.h

@@ -315,6 +315,8 @@ typedef struct SDeleteStmt {
   SNode*      pFromTable;  // FROM clause
   SNode*      pWhere;      // WHERE clause
   SNode*      pCountFunc;  // count the number of rows affected
+  SNode*      pFirstFunc;  // the start timestamp when the data was actually deleted
+  SNode*      pLastFunc;   // the end timestamp when the data was actually deleted
   SNode*      pTagCond;    // pWhere divided into pTagCond and timeRange
   STimeWindow timeRange;
   uint8_t     precision;

source/dnode/vnode/src/tq/tq.c

@@ -832,7 +832,7 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
   tDecoderClear(pCoder);
 
   int32_t sz = taosArrayGetSize(pRes->uidList);
-  if (sz == 0) {
+  if (sz == 0 || pRes->affectedRows == 0) {
     taosArrayDestroy(pRes->uidList);
     return 0;
   }

source/libs/executor/src/dataDeleter.c

@@ -79,25 +79,33 @@ static void toDataCacheEntry(SDataDeleterHandle* pHandle, const SInputData* pInp
   pEntry->dataLen = sizeof(SDeleterRes);
 
   ASSERT(1 == pEntry->numOfRows);
-  ASSERT(1 == pEntry->numOfCols);
+  ASSERT(3 == pEntry->numOfCols);
 
   pBuf->useSize = sizeof(SDataCacheEntry);
 
   SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pInput->pData->pDataBlock, 0);
+  SColumnInfoData* pColSKey = (SColumnInfoData*)taosArrayGet(pInput->pData->pDataBlock, 1);
+  SColumnInfoData* pColEKey = (SColumnInfoData*)taosArrayGet(pInput->pData->pDataBlock, 2);
 
   SDeleterRes* pRes = (SDeleterRes*)pEntry->data;
   pRes->suid = pHandle->pParam->suid;
   pRes->uidList = pHandle->pParam->pUidList;
-  pRes->skey = pHandle->pDeleter->deleteTimeRange.skey;
-  pRes->ekey = pHandle->pDeleter->deleteTimeRange.ekey;
   strcpy(pRes->tableName, pHandle->pDeleter->tableFName);
   strcpy(pRes->tsColName, pHandle->pDeleter->tsColName);
   pRes->affectedRows = *(int64_t*)pColRes->pData;
+  if (pRes->affectedRows) {
+    pRes->skey = *(int64_t*)pColSKey->pData;
+    pRes->ekey = *(int64_t*)pColEKey->pData;
+    ASSERT(pRes->skey <= pRes->ekey);
+  } else {
+    pRes->skey = pHandle->pDeleter->deleteTimeRange.skey;
+    pRes->ekey = pHandle->pDeleter->deleteTimeRange.ekey;
+  }
 
   pBuf->useSize += pEntry->dataLen;
-  
-  atomic_add_fetch_64(&pHandle->cachedSize, pEntry->dataLen); 
-  atomic_add_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen); 
+
+  atomic_add_fetch_64(&pHandle->cachedSize, pEntry->dataLen);
+  atomic_add_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);
 }
 
 static bool allocBuf(SDataDeleterHandle* pDeleter, const SInputData* pInput, SDataDeleterBuf* pBuf) {
@@ -172,7 +180,8 @@ static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, bool* pQueryE
   SDataCacheEntry* pEntry = (SDataCacheEntry*)pDeleter->nextOutput.pData;
   *pLen = pEntry->dataLen;
   *pQueryEnd = pDeleter->queryEnd;
-  qDebug("got data len %" PRId64 ", row num %d in sink", *pLen, ((SDataCacheEntry*)(pDeleter->nextOutput.pData))->numOfRows);
+  qDebug("got data len %" PRId64 ", row num %d in sink", *pLen,
+         ((SDataCacheEntry*)(pDeleter->nextOutput.pData))->numOfRows);
 }
 
 static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
@@ -186,14 +195,14 @@ static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
     return TSDB_CODE_SUCCESS;
   }
   SDataCacheEntry* pEntry = (SDataCacheEntry*)(pDeleter->nextOutput.pData);
-  memcpy(pOutput->pData, pEntry->data, pEntry->dataLen);  
+  memcpy(pOutput->pData, pEntry->data, pEntry->dataLen);
   pDeleter->pParam->pUidList = NULL;
   pOutput->numOfRows = pEntry->numOfRows;
   pOutput->numOfCols = pEntry->numOfCols;
   pOutput->compressed = pEntry->compressed;
 
-  atomic_sub_fetch_64(&pDeleter->cachedSize, pEntry->dataLen);  
-  atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen); 
+  atomic_sub_fetch_64(&pDeleter->cachedSize, pEntry->dataLen);
+  atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);
 
   taosMemoryFreeClear(pDeleter->nextOutput.pData);  // todo persistent
   pOutput->bufStatus = updateStatus(pDeleter);
@@ -202,7 +211,7 @@ static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
   pOutput->useconds = pDeleter->useconds;
   pOutput->precision = pDeleter->pSchema->precision;
   taosThreadMutexUnlock(&pDeleter->mutex);
-  
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -211,7 +220,7 @@ static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
   atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pDeleter->cachedSize);
   taosMemoryFreeClear(pDeleter->nextOutput.pData);
   taosArrayDestroy(pDeleter->pParam->pUidList);
-  taosMemoryFree(pDeleter->pParam);  
+  taosMemoryFree(pDeleter->pParam);
   while (!taosQueueEmpty(pDeleter->pDataBlocks)) {
     SDataDeleterBuf* pBuf = NULL;
     taosReadQitem(pDeleter->pDataBlocks, (void**)&pBuf);
@@ -230,14 +239,15 @@ static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t createDataDeleter(SDataSinkManager* pManager, const SDataSinkNode* pDataSink, DataSinkHandle* pHandle, void *pParam) {
+int32_t createDataDeleter(SDataSinkManager* pManager, const SDataSinkNode* pDataSink, DataSinkHandle* pHandle,
+                          void* pParam) {
   SDataDeleterHandle* deleter = taosMemoryCalloc(1, sizeof(SDataDeleterHandle));
   if (NULL == deleter) {
     terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
 
-  SDataDeleterNode* pDeleterNode = (SDataDeleterNode *)pDataSink;
+  SDataDeleterNode* pDeleterNode = (SDataDeleterNode*)pDataSink;
   deleter->sink.fPut = putDataBlock;
   deleter->sink.fEndPut = endPut;
   deleter->sink.fGetLen = getDataLength;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -399,6 +399,8 @@ static int32_t logicVnodeModifCopy(const SVnodeModifyLogicNode* pSrc, SVnodeModi
   COPY_SCALAR_FIELD(modifyType);
   COPY_SCALAR_FIELD(msgType);
   CLONE_NODE_FIELD(pAffectedRows);
+  CLONE_NODE_FIELD(pStartTs);
+  CLONE_NODE_FIELD(pEndTs);
   COPY_SCALAR_FIELD(tableId);
   COPY_SCALAR_FIELD(stableId);
   COPY_SCALAR_FIELD(tableType);

source/libs/nodes/src/nodesCodeFuncs.c

@@ -2431,6 +2431,8 @@ static const char* jkDeletePhysiPlanTsColName = "TsColName";
 static const char* jkDeletePhysiPlanDeleteTimeRangeStartKey = "DeleteTimeRangeStartKey";
 static const char* jkDeletePhysiPlanDeleteTimeRangeEndKey = "DeleteTimeRangeEndKey";
 static const char* jkDeletePhysiPlanAffectedRows = "AffectedRows";
+static const char* jkDeletePhysiPlanStartTs = "StartTs";
+static const char* jkDeletePhysiPlanEndTs = "EndTs";
 
 static int32_t physiDeleteNodeToJson(const void* pObj, SJson* pJson) {
   const SDataDeleterNode* pNode = (const SDataDeleterNode*)pObj;
@@ -2457,6 +2459,12 @@ static int32_t physiDeleteNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkDeletePhysiPlanAffectedRows, nodeToJson, pNode->pAffectedRows);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkDeletePhysiPlanStartTs, nodeToJson, pNode->pStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkDeletePhysiPlanEndTs, nodeToJson, pNode->pEndTs);
+  }
 
   return code;
 }
@@ -2486,6 +2494,12 @@ static int32_t jsonToPhysiDeleteNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeObject(pJson, jkDeletePhysiPlanAffectedRows, &pNode->pAffectedRows);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkDeletePhysiPlanStartTs, &pNode->pStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkDeletePhysiPlanEndTs, &pNode->pEndTs);
+  }
 
   return code;
 }

source/libs/nodes/src/nodesMsgFuncs.c

@@ -2665,7 +2665,9 @@ enum {
   PHY_DELETER_CODE_TABLE_FNAME,
   PHY_DELETER_CODE_TS_COL_NAME,
   PHY_DELETER_CODE_DELETE_TIME_RANGE,
-  PHY_DELETER_CODE_AFFECTED_ROWS
+  PHY_DELETER_CODE_AFFECTED_ROWS,
+  PHY_DELETER_CODE_START_TS,
+  PHY_DELETER_CODE_END_TS
 };
 
 static int32_t physiDeleteNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
@@ -2690,6 +2692,12 @@ static int32_t physiDeleteNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tlvEncodeObj(pEncoder, PHY_DELETER_CODE_AFFECTED_ROWS, nodeToMsg, pNode->pAffectedRows);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, PHY_DELETER_CODE_START_TS, nodeToMsg, pNode->pStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, PHY_DELETER_CODE_END_TS, nodeToMsg, pNode->pEndTs);
+  }
 
   return code;
 }
@@ -2722,6 +2730,12 @@ static int32_t msgToPhysiDeleteNode(STlvDecoder* pDecoder, void* pObj) {
       case PHY_DELETER_CODE_AFFECTED_ROWS:
         code = msgToNodeFromTlv(pTlv, (void**)&pNode->pAffectedRows);
         break;
+      case PHY_DELETER_CODE_START_TS:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pStartTs);
+        break;
+      case PHY_DELETER_CODE_END_TS:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pEndTs);
+        break;
       default:
         break;
     }

source/libs/nodes/src/nodesUtilFuncs.c

@@ -727,6 +727,8 @@ void nodesDestroyNode(SNode* pNode) {
       nodesDestroyNode(pStmt->pFromTable);
       nodesDestroyNode(pStmt->pWhere);
       nodesDestroyNode(pStmt->pCountFunc);
+      nodesDestroyNode(pStmt->pFirstFunc);
+      nodesDestroyNode(pStmt->pLastFunc);
       nodesDestroyNode(pStmt->pTagCond);
       break;
     }
@@ -791,6 +793,8 @@ void nodesDestroyNode(SNode* pNode) {
       destroyVgDataBlockArray(pLogicNode->pDataBlocks);
       // pVgDataBlocks is weak reference
       nodesDestroyNode(pLogicNode->pAffectedRows);
+      nodesDestroyNode(pLogicNode->pStartTs);
+      nodesDestroyNode(pLogicNode->pEndTs);
       taosMemoryFreeClear(pLogicNode->pVgroupList);
       nodesDestroyList(pLogicNode->pInsertCols);
       break;
@@ -997,6 +1001,8 @@ void nodesDestroyNode(SNode* pNode) {
       SDataDeleterNode* pSink = (SDataDeleterNode*)pNode;
       destroyDataSinkNode((SDataSinkNode*)pSink);
       nodesDestroyNode(pSink->pAffectedRows);
+      nodesDestroyNode(pSink->pStartTs);
+      nodesDestroyNode(pSink->pEndTs);
       break;
     }
     case QUERY_NODE_PHYSICAL_SUBPLAN: {

source/libs/parser/src/parAstCreater.c

@@ -1787,10 +1787,10 @@ SNode* createRevokeStmt(SAstCreateContext* pCxt, int64_t privileges, SToken* pDb
   return (SNode*)pStmt;
 }
 
-SNode* createCountFuncForDelete(SAstCreateContext* pCxt) {
+SNode* createFuncForDelete(SAstCreateContext* pCxt, const char* pFuncName) {
   SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
   CHECK_OUT_OF_MEM(pFunc);
-  strcpy(pFunc->functionName, "count");
+  strcpy(pFunc->functionName, pFuncName);
   if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pFunc->pParameterList, createPrimaryKeyCol(pCxt, NULL))) {
     nodesDestroyNode((SNode*)pFunc);
     CHECK_OUT_OF_MEM(NULL);
@@ -1804,8 +1804,10 @@ SNode* createDeleteStmt(SAstCreateContext* pCxt, SNode* pTable, SNode* pWhere) {
   CHECK_OUT_OF_MEM(pStmt);
   pStmt->pFromTable = pTable;
   pStmt->pWhere = pWhere;
-  pStmt->pCountFunc = createCountFuncForDelete(pCxt);
-  if (NULL == pStmt->pCountFunc) {
+  pStmt->pCountFunc = createFuncForDelete(pCxt, "count");
+  pStmt->pFirstFunc = createFuncForDelete(pCxt, "first");
+  pStmt->pLastFunc = createFuncForDelete(pCxt, "last");
+  if (NULL == pStmt->pCountFunc || NULL == pStmt->pFirstFunc || NULL == pStmt->pLastFunc) {
     nodesDestroyNode((SNode*)pStmt);
     CHECK_OUT_OF_MEM(NULL);
   }

source/libs/parser/src/parTranslater.c

@@ -3347,10 +3347,16 @@ static int32_t translateDelete(STranslateContext* pCxt, SDeleteStmt* pDelete) {
   if (TSDB_CODE_SUCCESS == code) {
     code = translateDeleteWhere(pCxt, pDelete);
   }
+  pCxt->currClause = SQL_CLAUSE_SELECT;
   if (TSDB_CODE_SUCCESS == code) {
-    pCxt->currClause = SQL_CLAUSE_SELECT;
     code = translateExpr(pCxt, &pDelete->pCountFunc);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = translateExpr(pCxt, &pDelete->pFirstFunc);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = translateExpr(pCxt, &pDelete->pLastFunc);
+  }
   return code;
 }
 

source/libs/planner/src/planLogicCreater.c

@@ -1372,9 +1372,21 @@ static int32_t createDeleteAggLogicNode(SLogicPlanContext* pCxt, SDeleteStmt* pD
   }
 
   int32_t code = nodesListMakeStrictAppend(&pAgg->pAggFuncs, nodesCloneNode(pDelete->pCountFunc));
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodesListStrictAppend(pAgg->pAggFuncs, nodesCloneNode(pDelete->pFirstFunc));
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodesListStrictAppend(pAgg->pAggFuncs, nodesCloneNode(pDelete->pLastFunc));
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = rewriteExpr(pAgg->pAggFuncs, &pDelete->pCountFunc);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteExpr(pAgg->pAggFuncs, &pDelete->pFirstFunc);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteExpr(pAgg->pAggFuncs, &pDelete->pLastFunc);
+  }
   // set the output
   if (TSDB_CODE_SUCCESS == code) {
     code = createColumnByRewriteExprs(pAgg->pAggFuncs, &pAgg->node.pTargets);
@@ -1405,7 +1417,9 @@ static int32_t createVnodeModifLogicNodeByDelete(SLogicPlanContext* pCxt, SDelet
   strcpy(pModify->tsColName, pRealTable->pMeta->schema->name);
   pModify->deleteTimeRange = pDelete->timeRange;
   pModify->pAffectedRows = nodesCloneNode(pDelete->pCountFunc);
-  if (NULL == pModify->pAffectedRows) {
+  pModify->pStartTs = nodesCloneNode(pDelete->pFirstFunc);
+  pModify->pEndTs = nodesCloneNode(pDelete->pLastFunc);
+  if (NULL == pModify->pAffectedRows || NULL == pModify->pStartTs || NULL == pModify->pEndTs) {
     nodesDestroyNode((SNode*)pModify);
     return TSDB_CODE_OUT_OF_MEMORY;
   }

source/libs/planner/src/planPhysiCreater.c

@@ -1323,9 +1323,9 @@ static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
 
 static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren,
                                         SPartitionLogicNode* pPartLogicNode, SPhysiNode** pPhyNode) {
-  SPartitionPhysiNode* pPart =
-      (SPartitionPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pPartLogicNode,
-          pCxt->pPlanCxt->streamQuery ? QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION : QUERY_NODE_PHYSICAL_PLAN_PARTITION);
+  SPartitionPhysiNode* pPart = (SPartitionPhysiNode*)makePhysiNode(
+      pCxt, (SLogicNode*)pPartLogicNode,
+      pCxt->pPlanCxt->streamQuery ? QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION : QUERY_NODE_PHYSICAL_PLAN_PARTITION);
   if (NULL == pPart) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
@@ -1670,6 +1670,12 @@ static int32_t createDataDeleter(SPhysiPlanContext* pCxt, SVnodeModifyLogicNode*
 
   int32_t code = setNodeSlotId(pCxt, pRoot->pOutputDataBlockDesc->dataBlockId, -1, pModify->pAffectedRows,
                                &pDeleter->pAffectedRows);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setNodeSlotId(pCxt, pRoot->pOutputDataBlockDesc->dataBlockId, -1, pModify->pStartTs, &pDeleter->pStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setNodeSlotId(pCxt, pRoot->pOutputDataBlockDesc->dataBlockId, -1, pModify->pEndTs, &pDeleter->pEndTs);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     pDeleter->sink.pInputDataBlockDesc = (SDataBlockDescNode*)nodesCloneNode((SNode*)pRoot->pOutputDataBlockDesc);
     if (NULL == pDeleter->sink.pInputDataBlockDesc) {