Feat/6550634959/external window

include/common/streamMsg.h

@@ -983,6 +983,7 @@ typedef struct SSTriggerCalcParam {
   int32_t notifyType;           // See also: ESTriggerEventType
   char*   extraNotifyContent;   // NULL if not available
   char*   resultNotifyContent;  // does not serialize
+  SArray* pExternalWindowData;
 } SSTriggerCalcParam;
 
 typedef struct SSTriggerCalcRequest {

include/libs/function/functionMgt.h

@@ -76,6 +76,7 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_TLCC,
   FUNCTION_TYPE_LAG,
   FUNCTION_TYPE_FILL_FORWARD,
+  FUNCTION_TYPE_EXTERNAL_WINDOW_COLUMN,
 
   // math function
   FUNCTION_TYPE_ABS = 1000,
@@ -356,6 +357,7 @@ bool fmIsRowTsOriginFunc(int32_t funcId);
 bool fmIsSelectColsFunc(int32_t funcId);
 bool fmIsGroupIdFunc(int32_t funcId);
 bool fmIsPlaceHolderFunc(int32_t funcId);
+bool fmIsPlaceHolderFuncForExternalWin(int32_t funcId);
 
 void    getLastCacheDataType(SDataType* pType, int32_t pkBytes);
 int32_t createFunction(const char* pName, SNodeList* pParameterList, SFunctionNode** pFunc);
@@ -394,6 +396,11 @@ int32_t fmGetFuncId(const char* name);
 bool    fmIsMyStateFunc(int32_t funcId, int32_t stateFuncId);
 bool    fmIsCountLikeFunc(int32_t funcId);
 
+int32_t fmGetTwstartFuncId();
+int32_t fmGetTwendFuncId();
+int32_t fmGetTwdurationFuncId();
+int32_t fmGetExternalWindowColumnFuncId();
+
 // typedef enum SStreamPseudoFuncType {
 //   STREAM_PSEUDO_FUNC_CURRENT_TS = 0,
 //   STREAM_PSEUDO_FUNC_TWSTART = 1,

include/libs/nodes/plannodes.h

@@ -391,6 +391,8 @@ typedef struct SWindowLogicNode {
   // for anomaly window
   SNode*                pAnomalyExpr;
   char                  anomalyOpt[TSDB_ANALYTIC_ALGO_OPTION_LEN];
+
+  SNode*                pSubquery;
 } SWindowLogicNode;
 
 typedef struct SFillLogicNode {
@@ -858,6 +860,7 @@ typedef struct SExternalWindowPhysiNode {
   bool             inputHasOrder;
   int32_t          orgTableVgId; // for vtable window query
   tb_uid_t         orgTableUid;  // for vtable window query
+  SNode*           pSubquery;
 } SExternalWindowPhysiNode;
 
 typedef struct SSortPhysiNode {

include/libs/nodes/querynodes.h

@@ -481,12 +481,15 @@ typedef struct SSlidingWindowNode {
 } SSlidingWindowNode;
 
 typedef struct SExternalWindowNode {
-  ENodeType   type;       // QUERY_NODE_EXTERNAL_WINDOW
+  ENodeType   type;  // QUERY_NODE_EXTERNAL_WINDOW
   SNodeList*  pProjectionList;
   SNodeList*  pAggFuncList;
   STimeWindow timeRange;
   SNode*      pTimeRange;
   void*       timezone;
+  SNode*      pSubquery;
+  SNode*      pFill;
+  char        aliasName[TSDB_COL_NAME_LEN];
 } SExternalWindowNode;
 
 typedef struct SStreamTriggerOptions {

include/libs/scalar/scalar.h

@@ -54,7 +54,7 @@ pDst need to freed in caller
 int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst, SScalarExtraInfo* pExtra);
 int32_t scalarCalculateInRange(SNode *pNode, SArray *pBlockList, SScalarParam *pDst, int32_t rowStartIdx, int32_t rowEndIdx, SScalarExtraInfo* pExtra);
 void    sclFreeParam(SScalarParam* param);
-int32_t scalarAssignPlaceHolderRes(SColumnInfoData* pResColData, int64_t offset, int64_t rows, int16_t funcId, const void* pExtraParams);
+int32_t scalarAssignPlaceHolderRes(SColumnInfoData* pResColData, int64_t offset, int64_t rows, int16_t funcId, const void* pExtraParams, SNode* pParamNode);
 int32_t scalarGetOperatorParamNum(EOperatorType type);
 int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type, STypeMod typeMod, int8_t processType);
 

source/libs/executor/src/externalwindowoperator.c

@@ -2242,6 +2242,7 @@ int32_t createExternalWindowOperator(SOperatorInfo* pDownstream, SPhysiNode* pNo
   QRY_PARAM_CHECK(pOptrOut);
   int32_t                  code = 0;
   int32_t                  lino = 0;
+  bool                     isInStream = true;
   SExternalWindowOperator* pExtW = taosMemoryCalloc(1, sizeof(SExternalWindowOperator));
   SOperatorInfo*           pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   pOperator->pPhyNode = pNode;
@@ -2271,6 +2272,65 @@ int32_t createExternalWindowOperator(SOperatorInfo* pDownstream, SPhysiNode* pNo
   // pExtW->limitInfo = (SLimitInfo){0};
   // initLimitInfo(pPhynode->window.node.pLimit, pPhynode->window.node.pSlimit, &pExtW->limitInfo);
 
+  if (pPhynode->pSubquery) {
+    // todo xs
+    // Initialize minimal stream runtime info to reuse external-window logic.
+    // Note: full subquery execution to populate multiple windows can be added later.
+    isInStream = false;
+    if (pTaskInfo->pStreamRuntimeInfo == NULL) {
+      pTaskInfo->pStreamRuntimeInfo = (SStreamRuntimeInfo*)taosMemoryCalloc(1, sizeof(SStreamRuntimeInfo));
+      if (pTaskInfo->pStreamRuntimeInfo == NULL) {
+        code = terrno;
+        goto _error;
+      }
+    }
+
+    SStreamRuntimeFuncInfo* pRt = &pTaskInfo->pStreamRuntimeInfo->funcInfo;
+    pRt->withExternalWindow = true;
+    pRt->isWindowTrigger = true;
+    pRt->triggerType = STREAM_TRIGGER_SESSION;
+    pRt->precision = 0;
+    pRt->curIdx = 0;
+    pRt->curOutIdx = 0;
+
+    if (pRt->pStreamPesudoFuncVals == NULL) {
+      pRt->pStreamPesudoFuncVals = taosArrayInit(4, sizeof(SSTriggerCalcParam));
+      if (pRt->pStreamPesudoFuncVals == NULL) {
+        code = terrno;
+        goto _error;
+      }
+    } else {
+      taosArrayClear(pRt->pStreamPesudoFuncVals);
+    }
+
+    SSTriggerCalcParam one = {0};
+    // Use provided timeRange as a minimal single-window placeholder
+    one.wstart = 1589335200000;
+    one.wend = 1589338140000;
+    one.wduration = one.wend - one.wstart;
+    SValue val = {0};
+    one.pExternalWindowData = taosArrayInit(1, sizeof(SValue));
+    QUERY_CHECK_NULL(one.pExternalWindowData, code, lino, _error, terrno);
+    val.type = TSDB_DATA_TYPE_BIGINT;
+    val.val = 10;
+    (void)taosArrayPush(one.pExternalWindowData, &val);
+    val.val = 11;
+    (void)taosArrayPush(one.pExternalWindowData, &val);
+
+    (void)taosArrayPush(pRt->pStreamPesudoFuncVals, &one);
+    one.wstart = 1589338140001;
+    one.wend = 1589340110000;
+    one.wduration = one.wend - one.wstart;
+    one.pExternalWindowData = taosArrayInit(1, sizeof(SValue));
+    QUERY_CHECK_NULL(one.pExternalWindowData, code, lino, _error, terrno);
+    val.type = TSDB_DATA_TYPE_BIGINT;
+    val.val = 20;
+    (void)taosArrayPush(one.pExternalWindowData, &val);
+    val.val = 21;
+    (void)taosArrayPush(one.pExternalWindowData, &val);
+    (void)taosArrayPush(pRt->pStreamPesudoFuncVals, &one);
+  }
+
   if (pPhynode->window.pProjs) {
     int32_t    numOfScalarExpr = 0;
     SExprInfo* pScalarExprInfo = NULL;
@@ -2387,11 +2447,29 @@ int32_t createExternalWindowOperator(SOperatorInfo* pDownstream, SPhysiNode* pNo
   }
 
   if (pPhynode->isSingleTable) {
-    pExtW->getWinFp = (pExtW->timeRangeExpr && (pExtW->timeRangeExpr->needCalc || (pTaskInfo->pStreamRuntimeInfo->funcInfo.addOptions & CALC_SLIDING_OVERLAP))) ? extWinGetOvlpWin : extWinGetNoOvlpWin;
-    pExtW->multiTableMode = false;
+    if (!isInStream) {
+      pExtW->getWinFp = extWinGetOvlpWin;
+      pExtW->multiTableMode = false;
+    } else {
+      pExtW->getWinFp =
+          (pExtW->timeRangeExpr && (pExtW->timeRangeExpr->needCalc ||
+                                    (pTaskInfo->pStreamRuntimeInfo->funcInfo.addOptions & CALC_SLIDING_OVERLAP)))
+              ? extWinGetOvlpWin
+              : extWinGetNoOvlpWin;
+      pExtW->multiTableMode = false;
+    }
   } else {
-    pExtW->getWinFp = (pExtW->timeRangeExpr && (pExtW->timeRangeExpr->needCalc || (pTaskInfo->pStreamRuntimeInfo->funcInfo.addOptions & CALC_SLIDING_OVERLAP))) ? extWinGetMultiTbOvlpWin : extWinGetMultiTbNoOvlpWin;
-    pExtW->multiTableMode = true;
+    if (!isInStream) {
+      pExtW->getWinFp = extWinGetMultiTbOvlpWin;
+      pExtW->multiTableMode = true;
+    } else {
+      pExtW->getWinFp =
+          (pExtW->timeRangeExpr && (pExtW->timeRangeExpr->needCalc ||
+                                    (pTaskInfo->pStreamRuntimeInfo->funcInfo.addOptions & CALC_SLIDING_OVERLAP)))
+              ? extWinGetMultiTbOvlpWin
+              : extWinGetMultiTbNoOvlpWin;
+      pExtW->multiTableMode = true;
+    }
   }
   pExtW->inputHasOrder = pPhynode->inputHasOrder;
   pExtW->orgTableUid = pPhynode->orgTableUid;

source/libs/executor/src/projectoperator.c

@@ -1014,7 +1014,8 @@ int32_t projectApplyFunction(SqlFunctionCtx* pCtx, SqlFunctionCtx* pfCtx, SExprI
   TSDB_CHECK_NULL(pResColData, code, lino, _exit, terrno);
 
   if (fmIsPlaceHolderFunc(pfCtx->functionId) && pExtraParams && pfCtx->pExpr->base.pParamList && 1 == pfCtx->pExpr->base.pParamList->length) {
-    TAOS_CHECK_EXIT(scalarAssignPlaceHolderRes(pResColData, pResult->info.rows, pSrcBlock->info.rows, pfCtx->functionId, pExtraParams));
+    SNode* pParamNode = nodesListGetNode(pfCtx->pExpr->base.pParamList, 0);
+    TAOS_CHECK_EXIT(scalarAssignPlaceHolderRes(pResColData, pResult->info.rows, pSrcBlock->info.rows, pfCtx->functionId, pExtraParams, pParamNode));
     *numOfRows = pSrcBlock->info.rows;
 
     return code;

source/libs/function/src/builtins.c

@@ -1203,6 +1203,10 @@ static int32_t translatePlaceHolderPseudoColumn(SFunctionNode* pFunc, char* pErr
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateExternalWindowColumnFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t translateTimePseudoColumn(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   // pseudo column do not need to check parameters
 
@@ -7243,6 +7247,20 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .estimateReturnRowsFunc = fillforwardEstReturnRows,
     .processFuncByRow  = fillforwardFunctionByRow,
   },
+  {
+    .name = "_external_window_column",
+    .type = FUNCTION_TYPE_EXTERNAL_WINDOW_COLUMN,
+    .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_PLACE_HOLDER_FUNC | FUNC_MGT_SKIP_SCAN_CHECK_FUNC,
+    .parameters = {.minParamNum = 0,
+                   .maxParamNum = 0,
+                   .paramInfoPattern = 0,
+                   .outputParaInfo = {.validDataType = FUNC_PARAM_SUPPORT_ALL_TYPE},},
+    .translateFunc = translateExternalWindowColumnFunc,
+    .getEnvFunc   = NULL,
+    .initFunc     = NULL,
+    .sprocessFunc = streamPseudoScalarFunction,
+    .finalizeFunc = NULL,
+  },
 };
 // clang-format on
 

source/libs/function/src/functionMgt.c

@@ -33,6 +33,14 @@ static SFuncMgtService gFunMgtService;
 static TdThreadOnce    functionHashTableInit = PTHREAD_ONCE_INIT;
 static int32_t         initFunctionCode = 0;
 
+static void InitSpecialFuncId() {
+  // just to make sure the funcId of these functions are initialized before used.
+  int32_t tmp = fmGetTwstartFuncId();
+  tmp = fmGetTwendFuncId();
+  tmp = fmGetTwdurationFuncId();
+  tmp = fmGetExternalWindowColumnFuncId();
+}
+
 static void doInitFunctionTable() {
   gFunMgtService.pFuncNameHashTable =
       taosHashInit(funcMgtBuiltinsNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
@@ -48,6 +56,7 @@ static void doInitFunctionTable() {
       return;
     }
   }
+  InitSpecialFuncId();
 }
 
 static bool isSpecificClassifyFunc(int32_t funcId, uint64_t classification) {
@@ -416,6 +425,11 @@ bool fmIsPrimaryKeyFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId,
 
 bool fmIsPlaceHolderFunc(int32_t funcId) {return isSpecificClassifyFunc(funcId, FUNC_MGT_PLACE_HOLDER_FUNC); }
 
+bool fmIsPlaceHolderFuncForExternalWin(int32_t funcId) {
+  return (funcMgtBuiltins[funcId].type == FUNCTION_TYPE_WSTART || funcMgtBuiltins[funcId].type == FUNCTION_TYPE_WEND ||
+          funcMgtBuiltins[funcId].type == FUNCTION_TYPE_WDURATION);
+}
+
 void getLastCacheDataType(SDataType* pType, int32_t pkBytes) {
   // TODO: do it later.
   pType->bytes = getFirstLastInfoSize(pType->bytes, pkBytes) + VARSTR_HEADER_SIZE;
@@ -720,6 +734,38 @@ int32_t fmGetFuncId(const char* name) {
   return -1;
 }
 
+int32_t fmGetTwstartFuncId() {
+  static int32_t twstartFuncId = -2;
+  if (twstartFuncId < 0) {
+    twstartFuncId = fmGetFuncId("_twstart");
+  }
+  return twstartFuncId;
+}
+
+int32_t fmGetTwendFuncId() {
+  static int32_t twendFuncId = -2;
+  if (twendFuncId < 0) {
+    twendFuncId = fmGetFuncId("_twend");
+  }
+  return twendFuncId;
+}
+
+int32_t fmGetTwdurationFuncId() {
+  static int32_t twdurationFuncId = -2;
+  if (twdurationFuncId < 0) {
+    twdurationFuncId = fmGetFuncId("_twduration");
+  }
+  return twdurationFuncId;
+}
+
+int32_t fmGetExternalWindowColumnFuncId() {
+  static int32_t externalWindowColumnFuncId = -2;
+  if (externalWindowColumnFuncId < 0) {
+    externalWindowColumnFuncId = fmGetFuncId("_external_window_column");
+  }
+  return externalWindowColumnFuncId;
+}
+
 bool fmIsMyStateFunc(int32_t funcId, int32_t stateFuncId) {
   const SBuiltinFuncDefinition* pFunc = &funcMgtBuiltins[funcId];
   const SBuiltinFuncDefinition* pStateFunc = &funcMgtBuiltins[stateFuncId];
@@ -753,6 +799,11 @@ bool fmIsGroupIdFunc(int32_t funcId) {
   return FUNCTION_TYPE_GROUP_ID == funcMgtBuiltins[funcId].type;
 }
 
+const void* fmGetExternalWindowColumnFuncVal(const SStreamRuntimeFuncInfo* pStreamRuntimeFuncInfo, int32_t index) {
+  SSTriggerCalcParam *pParams = taosArrayGet(pStreamRuntimeFuncInfo->pStreamPesudoFuncVals, pStreamRuntimeFuncInfo->curIdx);
+  return taosArrayGet(pParams->pExternalWindowData, index);
+}
+
 const void* fmGetStreamPesudoFuncVal(int32_t funcId, const SStreamRuntimeFuncInfo* pStreamRuntimeFuncInfo) {
   SSTriggerCalcParam *pParams = taosArrayGet(pStreamRuntimeFuncInfo->pStreamPesudoFuncVals, pStreamRuntimeFuncInfo->curIdx);
   switch (funcMgtBuiltins[funcId].type) {
@@ -889,6 +940,19 @@ int32_t fmSetStreamPseudoFuncParamVal(int32_t funcId, SNodeList* pParamNodes, co
         break;
       }
     }
+  } else if(FUNCTION_TYPE_EXTERNAL_WINDOW_COLUMN == t) {
+    // todo xs external_window_column
+    SNode* pParamNode = nodesListGetNode(pParamNodes, 0);
+    const void* pVal = fmGetExternalWindowColumnFuncVal(pStreamRuntimeInfo, ((SValueNode*)pParamNode)->placeholderNo);
+    if (!pVal) {
+      uError("failed to set stream pseudo func param val, NULL val for funcId: %d", funcId);
+      return TSDB_CODE_INTERNAL_ERROR;
+    }
+    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, &((SValue*)pVal)->val);
+    if (code != 0) {
+      uError("failed to set value node value: %s", tstrerror(code));
+      return code;
+    }
   } else if (LIST_LENGTH(pParamNodes) == 1) {
     // twstart, twend
     const void* pVal = fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);

source/libs/nodes/src/nodesCloneFuncs.c

@@ -759,6 +759,7 @@ static int32_t logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* p
   COPY_SCALAR_FIELD(extendOption);
   COPY_SCALAR_FIELD(orgTableUid);
   COPY_SCALAR_FIELD(orgTableVgId);
+  CLONE_NODE_FIELD(pSubquery);
   return TSDB_CODE_SUCCESS;
 }