fix: the problem of data loss when interval is used for outer query

include/libs/function/functionMgt.h

@@ -197,6 +197,7 @@ bool fmIsSystemInfoFunc(int32_t funcId);
 bool fmIsImplicitTsFunc(int32_t funcId);
 bool fmIsClientPseudoColumnFunc(int32_t funcId);
 bool fmIsMultiRowsFunc(int32_t funcId);
+bool fmIsKeepOrderFunc(int32_t funcId);
 
 int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
 

source/libs/function/inc/functionMgtInt.h

@@ -47,6 +47,7 @@ extern "C" {
 #define FUNC_MGT_SYSTEM_INFO_FUNC       FUNC_MGT_FUNC_CLASSIFICATION_MASK(18)
 #define FUNC_MGT_CLIENT_PC_FUNC         FUNC_MGT_FUNC_CLASSIFICATION_MASK(19)
 #define FUNC_MGT_MULTI_ROWS_FUNC        FUNC_MGT_FUNC_CLASSIFICATION_MASK(20)
+#define FUNC_MGT_KEEP_ORDER_FUNC        FUNC_MGT_FUNC_CLASSIFICATION_MASK(21)
 
 #define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 

source/libs/function/src/builtins.c

@@ -2097,7 +2097,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "top",
     .type = FUNCTION_TYPE_TOP,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateTopBot,
     .getEnvFunc   = getTopBotFuncEnv,
     .initFunc     = topBotFunctionSetup,
@@ -2112,7 +2112,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "bottom",
     .type = FUNCTION_TYPE_BOTTOM,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateTopBot,
     .getEnvFunc   = getTopBotFuncEnv,
     .initFunc     = topBotFunctionSetup,
@@ -2480,7 +2480,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "sample",
     .type = FUNCTION_TYPE_SAMPLE,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateSample,
     .getEnvFunc   = getSampleFuncEnv,
     .initFunc     = sampleFunctionSetup,
@@ -2906,7 +2906,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "_select_value",
     .type = FUNCTION_TYPE_SELECT_VALUE,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_KEEP_ORDER_FUNC,
     .translateFunc = translateSelectValue,
     .getEnvFunc   = getSelectivityFuncEnv,  // todo remove this function later.
     .initFunc     = functionSetup,

source/libs/function/src/functionMgt.c

@@ -183,6 +183,8 @@ bool fmIsClientPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(
 
 bool fmIsMultiRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_ROWS_FUNC); }
 
+bool fmIsKeepOrderFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_KEEP_ORDER_FUNC); }
+
 bool fmIsInterpFunc(int32_t funcId) {
   if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
     return false;

source/libs/planner/src/planOptimizer.c

@@ -2010,15 +2010,16 @@ static int32_t rewriteUniqueOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan*
   }
   if (TSDB_CODE_SUCCESS == code) {
     code = nodesListMakeAppend(&pProject->pChildren, (SNode*)pAgg);
-    pAgg->pParent = pProject;
-    pAgg = NULL;
   }
   if (TSDB_CODE_SUCCESS == code) {
+    pAgg->pParent = pProject;
+    pAgg = NULL;
     code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pIndef, pProject);
   }
   if (TSDB_CODE_SUCCESS == code) {
     code = adjustLogicNodeDataRequirement(
         pProject, NULL == pProject->pParent ? DATA_ORDER_LEVEL_NONE : pProject->pParent->requireDataOrder);
+    pProject = NULL;
   }
   if (TSDB_CODE_SUCCESS == code) {
     nodesDestroyNode((SNode*)pIndef);

source/libs/planner/src/planUtil.c

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "functionMgt.h"
 #include "planInt.h"
 
 static char* getUsageErrFormat(int32_t errCode) {
@@ -140,13 +141,27 @@ static int32_t adjustScanDataRequirement(SScanLogicNode* pScan, EDataOrderLevel
 }
 
 static int32_t adjustJoinDataRequirement(SJoinLogicNode* pJoin, EDataOrderLevel requirement) {
+  // The lowest sort level of join input and output data is DATA_ORDER_LEVEL_GLOBAL
   return TSDB_CODE_SUCCESS;
 }
 
+static bool isKeepOrderAggFunc(SNodeList* pFuncs) {
+  SNode* pFunc = NULL;
+  FOREACH(pFunc, pFuncs) {
+    if (!fmIsKeepOrderFunc(((SFunctionNode*)pFunc)->funcId)) {
+      return false;
+    }
+  }
+  return true;
+}
+
 static int32_t adjustAggDataRequirement(SAggLogicNode* pAgg, EDataOrderLevel requirement) {
-  if (requirement > DATA_ORDER_LEVEL_NONE) {
+  // The sort level of agg with group by output data can only be DATA_ORDER_LEVEL_NONE
+  if (requirement > DATA_ORDER_LEVEL_NONE && (NULL != pAgg->pGroupKeys || !isKeepOrderAggFunc(pAgg->pAggFuncs))) {
     return TSDB_CODE_PLAN_INTERNAL_ERROR;
   }
+  pAgg->node.resultDataOrder = requirement;
+  pAgg->node.requireDataOrder = requirement;
   return TSDB_CODE_SUCCESS;
 }
 
@@ -157,11 +172,12 @@ static int32_t adjustProjectDataRequirement(SProjectLogicNode* pProject, EDataOr
 }
 
 static int32_t adjustIntervalDataRequirement(SWindowLogicNode* pWindow, EDataOrderLevel requirement) {
-  if (requirement <= pWindow->node.resultDataOrder) {
-    return TSDB_CODE_SUCCESS;
+  // The lowest sort level of interval output data is DATA_ORDER_LEVEL_IN_GROUP
+  if (requirement < DATA_ORDER_LEVEL_IN_GROUP) {
+    requirement = DATA_ORDER_LEVEL_IN_GROUP;
   }
+  // The sort level of interval input data is always DATA_ORDER_LEVEL_IN_BLOCK
   pWindow->node.resultDataOrder = requirement;
-  pWindow->node.requireDataOrder = requirement;
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/planner/test/planSubqueryTest.cpp

@@ -70,4 +70,6 @@ TEST_F(PlanSubqeuryTest, outerInterval) {
   run("SELECT COUNT(*) FROM (SELECT * FROM st1) INTERVAL(5s)");
 
   run("SELECT COUNT(*) + SUM(c1) FROM (SELECT * FROM st1) INTERVAL(5s)");
+
+  run("SELECT COUNT(*) FROM (SELECT ts, TOP(c1, 10) FROM st1s1) INTERVAL(5s)");
 }