enh(query): time window interpolation for interval query.

source/libs/executor/inc/executorInt.h

@@ -20,6 +20,12 @@
 extern "C" {
 #endif
 
+typedef struct {
+  char*           pData;
+  bool            isNull;
+  int16_t         type;
+  int32_t         bytes;
+} SGroupKeys, SStateKeys;
 
 #ifdef __cplusplus
 }

source/libs/executor/inc/executorimpl.h

@@ -40,6 +40,7 @@ extern "C" {
 #include "tpagedbuf.h"
 
 #include "vnode.h"
+#include "executorInt.h"
 
 typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int32_t order);
 
@@ -478,16 +479,8 @@ typedef struct SFillOperatorInfo {
   void**            p;
   SSDataBlock*      existNewGroupBlock;
   bool              multigroupResult;
-  SInterval         intervalInfo;
 } SFillOperatorInfo;
 
-typedef struct {
-  char*           pData;
-  bool            isNull;
-  int16_t         type;
-  int32_t         bytes;
-} SGroupKeys, SStateKeys;
-
 typedef struct SGroupbyOperatorInfo {
   SOptrBasicInfo  binfo;
   SArray*         pGroupCols;     // group by columns, SArray<SColumn>
@@ -676,7 +669,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
                                             SArray* pTableIdList, SExecTaskInfo* pTaskInfo, SNode* pConditions);
 
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
-                                      SInterval* pInterval, SSDataBlock* pResBlock, int32_t fillType, char* fillVal,
+                                      SInterval* pInterval, STimeWindow* pWindow, SSDataBlock* pResBlock, int32_t fillType, SNodeListNode* fillVal,
                                       bool multigroupResult, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
                                              SSDataBlock* pResBlock, STimeWindowAggSupp *pTwAggSupp, SExecTaskInfo* pTaskInfo);

source/libs/executor/inc/tfill.h

@@ -27,13 +27,12 @@ extern "C" {
 struct SSDataBlock;
 
 typedef struct SFillColInfo {
-//  STColumn col;             // column info
-  SResSchema col;
-  int16_t  functionId;      // sql function id
-  int16_t  flag;            // column flag: TAG COLUMN|NORMAL COLUMN
-  int16_t  tagIndex;        // index of current tag in SFillTagColInfo array list
-  int32_t  offset;
-  union {int64_t i; double d;} val;
+  SExprInfo *pExpr;
+//  SResSchema schema;
+//  int16_t    functionId;      // sql function id
+  int16_t    flag;            // column flag: TAG COLUMN|NORMAL COLUMN
+  int16_t    tagIndex;        // index of current tag in SFillTagColInfo array list
+  SVariant   fillVal;
 } SFillColInfo;
 
 typedef struct {
@@ -56,9 +55,10 @@ typedef struct SFillInfo {
   int32_t   numOfCols;            // number of columns, including the tags columns
   int32_t   rowSize;              // size of each row
   SInterval interval;
-  char *    prevValues;           // previous row of data, to generate the interpolation results
-  char *    nextValues;           // next row of data
-  char**    pData;                // original result data block involved in filling data
+
+  SArray   *prev;
+  SArray   *next;
+  SSDataBlock *pSrcBlock;
   int32_t   alloc;                // data buffer size in rows
 
   SFillColInfo* pFillCol;         // column info for fill operations
@@ -72,7 +72,7 @@ int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t
 void taosFillSetStartInfo(struct SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey);
 void taosResetFillInfo(struct SFillInfo* pFillInfo, TSKEY startTimestamp);
 void taosFillSetInputDataBlock(struct SFillInfo* pFillInfo, const struct SSDataBlock* pInput);
-struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, const struct SValueNode* val);
+struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, const struct SNodeListNode* val);
 bool taosFillHasMoreResults(struct SFillInfo* pFillInfo);
 
 SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
@@ -80,7 +80,7 @@ SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int3
                                      struct SFillColInfo* pCol, const char* id);
 
 void* taosDestroyFillInfo(struct SFillInfo *pFillInfo);
-int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, void** output, int32_t capacity);
+int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity);
 int64_t getFillInfoStart(struct SFillInfo *pFillInfo);
 
 

source/libs/executor/src/executorimpl.c

@@ -3201,16 +3201,16 @@ static int32_t compressQueryColData(SColumnInfoData* pColRes, int32_t numOfRows,
                                                       colSize + COMP_OVERFLOW_BYTES, compressed, NULL, 0);
 }
 
-int32_t doFillTimeIntervalGapsInResults(struct SFillInfo* pFillInfo, SSDataBlock* pOutput, int32_t capacity, void** p) {
+int32_t doFillTimeIntervalGapsInResults(struct SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t capacity) {
   //  for(int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
   //    SColumnInfoData* pColInfoData = taosArrayGet(pOutput->pDataBlock, i);
   //    p[i] = pColInfoData->pData + (pColInfoData->info.bytes * pOutput->info.rows);
   //  }
 
-  int32_t numOfRows = (int32_t)taosFillResultDataBlock(pFillInfo, p, capacity - pOutput->info.rows);
-  pOutput->info.rows += numOfRows;
+  int32_t numOfRows = (int32_t)taosFillResultDataBlock(pFillInfo, pBlock, capacity - pBlock->info.rows);
+  pBlock->info.rows += numOfRows;
 
-  return pOutput->info.rows;
+  return pBlock->info.rows;
 }
 
 void publishOperatorProfEvent(SOperatorInfo* pOperator, EQueryProfEventType eventType) {
@@ -5559,7 +5559,7 @@ static void doHandleRemainBlockForNewGroupImpl(SFillOperatorInfo* pInfo, SResult
   taosFillSetStartInfo(pInfo->pFillInfo, pInfo->existNewGroupBlock->info.rows, ekey);
   taosFillSetInputDataBlock(pInfo->pFillInfo, pInfo->existNewGroupBlock);
 
-  doFillTimeIntervalGapsInResults(pInfo->pFillInfo, pInfo->pRes, pResultInfo->capacity, pInfo->p);
+  doFillTimeIntervalGapsInResults(pInfo->pFillInfo, pInfo->pRes, pResultInfo->capacity);
   pInfo->existNewGroupBlock = NULL;
   *newgroup = true;
 }
@@ -5568,7 +5568,7 @@ static void doHandleRemainBlockFromNewGroup(SFillOperatorInfo* pInfo, SResultInf
                                             SExecTaskInfo* pTaskInfo) {
   if (taosFillHasMoreResults(pInfo->pFillInfo)) {
     *newgroup = false;
-    doFillTimeIntervalGapsInResults(pInfo->pFillInfo, pInfo->pRes, (int32_t)pResultInfo->capacity, pInfo->p);
+    doFillTimeIntervalGapsInResults(pInfo->pFillInfo, pInfo->pRes, (int32_t)pResultInfo->capacity);
     if (pInfo->pRes->info.rows > pResultInfo->threshold || (!pInfo->multigroupResult)) {
       return;
     }
@@ -5629,7 +5629,8 @@ static SSDataBlock* doFill(SOperatorInfo* pOperator, bool* newgroup) {
       }
     }
 
-    doFillTimeIntervalGapsInResults(pInfo->pFillInfo, pResBlock, pOperator->resultInfo.capacity, pInfo->p);
+    blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);
+    doFillTimeIntervalGapsInResults(pInfo->pFillInfo, pResBlock, pOperator->resultInfo.capacity);
 
     // current group has no more result to return
     if (pResBlock->info.rows > 0) {
@@ -6203,19 +6204,17 @@ _error:
   return NULL;
 }
 
-static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t numOfCols, int64_t* fillVal,
+static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t numOfCols, SNodeListNode* pValNode,
                             STimeWindow win, int32_t capacity, const char* id, SInterval* pInterval, int32_t fillType) {
-  SFillColInfo* pColInfo = createFillColInfo(pExpr, numOfCols, NULL);
+  SFillColInfo* pColInfo = createFillColInfo(pExpr, numOfCols, pValNode);
 
-  // TODO set correct time precision
   STimeWindow w = TSWINDOW_INITIALIZER;
-  getAlignQueryTimeWindow(pInterval, TSDB_TIME_PRECISION_MILLI, win.skey, &w);
+  getAlignQueryTimeWindow(pInterval, pInterval->precision, win.skey, &w);
 
   int32_t order = TSDB_ORDER_ASC;
   pInfo->pFillInfo = taosCreateFillInfo(order, w.skey, 0, capacity, numOfCols, pInterval, fillType, pColInfo, id);
 
   pInfo->p = taosMemoryCalloc(numOfCols, POINTER_BYTES);
-
   if (pInfo->pFillInfo == NULL || pInfo->p == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;
   } else {
@@ -6223,15 +6222,29 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t
   }
 }
 
+static SArray* getFillValue(SNodeListNode* pNodeList) {
+  SArray* pList = taosArrayInit(4, sizeof(SVariant));
+
+  size_t  len = LIST_LENGTH(pNodeList->pNodeList);
+  for(int32_t i = 0; i < len; ++i) {
+    SValueNode* pvalue = (SValueNode*)nodesListGetNode(pNodeList->pNodeList, i);
+
+    SVariant v = {0};
+    valueNodeToVariant(pvalue, &v);
+    taosArrayPush(pList, &v);
+  }
+
+  return pList;
+}
+
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
-                                      SInterval* pInterval, SSDataBlock* pResBlock, int32_t fillType, char* fillVal,
+                                      SInterval* pInterval, STimeWindow* pWindow, SSDataBlock* pResBlock, int32_t fillType, SNodeListNode* pValueNode,
                                       bool multigroupResult, SExecTaskInfo* pTaskInfo) {
   SFillOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SFillOperatorInfo));
   SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
 
   pInfo->pRes = pResBlock;
   pInfo->multigroupResult = multigroupResult;
-  pInfo->intervalInfo = *pInterval;
 
   int32_t type = TSDB_FILL_NONE;
   switch (fillType) {
@@ -6260,7 +6273,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExp
   SResultInfo* pResultInfo = &pOperator->resultInfo;
   initResultSizeInfo(pOperator, 4096);
 
-  int32_t code = initFillInfo(pInfo, pExpr, numOfCols, (int64_t*)fillVal, pTaskInfo->window, pResultInfo->capacity,
+  int32_t code = initFillInfo(pInfo, pExpr, numOfCols, pValueNode, *pWindow, pResultInfo->capacity,
                               pTaskInfo->id.str, pInterval, type);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
@@ -6269,7 +6282,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExp
   pOperator->name = "FillOperator";
   pOperator->blockingOptr = false;
   pOperator->status = OP_NOT_OPENED;
-  //  pOperator->operatorType = OP_Fill;
+  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL;
   pOperator->pExpr = pExpr;
   pOperator->numOfOutput = numOfCols;
   pOperator->info = pInfo;
@@ -6618,13 +6631,6 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     int32_t primaryTsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
     pOptr = createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, primaryTsSlotId, &as,
                                        pTableGroupInfo, pTaskInfo);
-
-    // if (pIntervalPhyNode->pFill != NULL) {
-    //   pOptr = createFillOperatorInfo(pOptr, pExprInfo, num, &interval, pResBlock, pIntervalPhyNode->pFill->mode,
-    //   NULL,
-    //                                  false, pTaskInfo);
-    // }
-
   } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
     SSortPhysiNode* pSortPhyNode = (SSortPhysiNode*)pPhyNode;
 
@@ -6662,6 +6668,13 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
 
     SExprInfo* pExprInfo = createExprInfo(pJoinNode->pTargets, NULL, &num);
     pOptr = createJoinOperatorInfo(ops, size, pExprInfo, num, pResBlock, pJoinNode->pOnConditions, pTaskInfo);
+  } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) {
+    SFillPhysiNode* pFillNode = (SFillPhysiNode*)pPhyNode;
+    SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
+    SExprInfo* pExprInfo = createExprInfo(pFillNode->pTargets, NULL, &num);
+
+    SInterval* pInterval = &((STableIntervalOperatorInfo*)ops[0]->info)->interval;
+    pOptr = createFillOperatorInfo(ops[0], pExprInfo, num, pInterval, &pFillNode->timeRange, pResBlock, pFillNode->mode, (SNodeListNode*)pFillNode->pValues, false, pTaskInfo);
   } else {
     ASSERT(0);
   }

source/libs/nodes/src/nodesCodeFuncs.c

@@ -1521,7 +1521,7 @@ static int32_t physiFillNodeToJson(const void* pObj, SJson* pJson) {
 static int32_t jsonToPhysiFillNode(const SJson* pJson, void* pObj) {
   SFillPhysiNode* pNode = (SFillPhysiNode*)pObj;
 
-  int32_t code = jsonToPhysiWindowNode(pJson, pObj);
+  int32_t code = jsonToPhysicPlanNode(pJson, pObj);
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetNumberValue(pJson, jkFillPhysiPlanMode, pNode->mode);
   }

source/libs/qworker/src/qworker.c

@@ -945,7 +945,8 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t ex
 
   code = qStringToSubplan(qwMsg->msg, &plan);
   if (TSDB_CODE_SUCCESS != code) {
-    QW_TASK_ELOG("task string to subplan failed, code:%x - %s", code, tstrerror(code));
+    code = TSDB_CODE_INVALID_MSG;
+    QW_TASK_ELOG("task physical plan to subplan failed, code:%x - %s", code, tstrerror(code));
     QW_ERR_JRET(code);
   }