Merge pull request #15867 from taosdata/fix/TD-18115

fix(query): interp + fill(linear) not working

source/libs/executor/inc/executorimpl.h

@@ -732,18 +732,19 @@ typedef struct SStreamSessionAggOperatorInfo {
 } SStreamSessionAggOperatorInfo;
 
 typedef struct STimeSliceOperatorInfo {
-  SSDataBlock*   pRes;
-  STimeWindow    win;
-  SInterval      interval;
-  int64_t        current;
-  SArray*        pPrevRow;      // SArray<SGroupValue>
-  SArray*        pNextRow;      // SArray<SGroupValue>
-  bool           isPrevRowSet;
-  bool           isNextRowSet;
-  int32_t        fillType;      // fill type
-  SColumn        tsCol;         // primary timestamp column
-  SExprSupp      scalarSup;     // scalar calculation
-  struct SFillColInfo*  pFillColInfo;  // fill column info
+  SSDataBlock*            pRes;
+  STimeWindow             win;
+  SInterval               interval;
+  int64_t                 current;
+  SArray*                 pPrevRow;      // SArray<SGroupValue>
+  SArray*                 pNextRow;      // SArray<SGroupValue>
+  SArray*                 pLinearInfo;   // SArray<SFillLinearInfo>
+  bool                    isPrevRowSet;
+  bool                    isNextRowSet;
+  int32_t                 fillType;      // fill type
+  SColumn                 tsCol;         // primary timestamp column
+  SExprSupp               scalarSup;     // scalar calculation
+  struct SFillColInfo*    pFillColInfo;  // fill column info
 } STimeSliceOperatorInfo;
 
 typedef struct SStateWindowOperatorInfo {

source/libs/executor/inc/tfill.h

@@ -33,11 +33,20 @@ typedef struct SFillColInfo {
   SVariant   fillVal;
 } SFillColInfo;
 
+typedef struct SFillLinearInfo {
+  SPoint  start;
+  SPoint  end;
+  bool    hasNull;
+  bool    fillLastPoint;
+  int16_t type;
+  int32_t bytes;
+} SFillLinearInfo;
+
 typedef struct {
   SSchema col;
   char*   tagVal;
 } SFillTagColInfo;
-  
+
 typedef struct SFillInfo {
   TSKEY     start;                // start timestamp
   TSKEY     end;                  // endKey for fill

source/libs/executor/src/tfill.c

@@ -669,4 +669,4 @@ SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, const str
   }
 
   return pFillCol;
-}
\ No newline at end of file
+}

source/libs/executor/src/timewindowoperator.c

@@ -2087,6 +2087,34 @@ static void doKeepNextRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock
   pSliceInfo->isNextRowSet = true;
 }
 
+static void doKeepLinearInfo(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock* pBlock, int32_t rowIndex) {
+  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
+    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
+    SFillLinearInfo* pLinearInfo = taosArrayGet(pSliceInfo->pLinearInfo, i);
+
+    // null data should not be kept since it can not be used to perform interpolation
+    if (!colDataIsNull_s(pColInfoData, i)) {
+      int64_t startKey = *(int64_t*)colDataGetData(pTsCol, rowIndex);
+      int64_t endKey   = *(int64_t*)colDataGetData(pTsCol, rowIndex + 1);
+      pLinearInfo->start.key = startKey;
+      pLinearInfo->end.key   = endKey;
+
+      char* val;
+      val = colDataGetData(pColInfoData, rowIndex);
+      memcpy(pLinearInfo->start.val, val, pLinearInfo->bytes);
+      val = colDataGetData(pColInfoData, rowIndex + 1);
+      memcpy(pLinearInfo->end.val, val, pLinearInfo->bytes);
+
+      pLinearInfo->hasNull = false;
+    } else {
+      pLinearInfo->hasNull = true;
+    }
+  }
+
+}
+
 static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pBlock,
                                    SSDataBlock* pResBlock) {
   int32_t rows = pResBlock->info.rows;
@@ -2115,52 +2143,41 @@ static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
         if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
           float v = 0;
           GET_TYPED_DATA(v, float, pVar->nType, &pVar->i);
-          colDataAppend(pDst, rows, (char*)&v, false);
+          colDataAppend(pDst, rows, (char *)&v, false);
         } else if (pDst->info.type == TSDB_DATA_TYPE_DOUBLE) {
           double v = 0;
           GET_TYPED_DATA(v, double, pVar->nType, &pVar->i);
-          colDataAppend(pDst, rows, (char*)&v, false);
+          colDataAppend(pDst, rows, (char *)&v, false);
         } else if (IS_SIGNED_NUMERIC_TYPE(pDst->info.type)) {
           int64_t v = 0;
           GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);
-          colDataAppend(pDst, rows, (char*)&v, false);
+          colDataAppend(pDst, rows, (char *)&v, false);
         }
         pResBlock->info.rows += 1;
         break;
       }
 
       case TSDB_FILL_LINEAR: {
-#if 0
-        if (pCtx->start.key == INT64_MIN || pCtx->start.key > pCtx->startTs
-                    || pCtx->end.key == INT64_MIN || pCtx->end.key < pCtx->startTs) {
-//                  goto interp_exit;
-                }
+        SFillLinearInfo* pLinearInfo = taosArrayGet(pSliceInfo->pLinearInfo, srcSlot);
 
-              double v1 = -1, v2 = -1;
-              GET_TYPED_DATA(v1, double, pCtx->inputType, &pCtx->start.val);
-              GET_TYPED_DATA(v2, double, pCtx->inputType, &pCtx->end.val);
+        SPoint start   = pLinearInfo->start;
+        SPoint end     = pLinearInfo->end;
+        SPoint current = {.key = pSliceInfo->current};
+        current.val = taosMemoryCalloc(pLinearInfo->bytes, 1);
 
-              SPoint point1 = {.key = ts, .val = &v1};
-              SPoint point2 = {.key = nextTs, .val = &v2};
-              SPoint point  = {.key = pCtx->startTs, .val = pCtx->pOutput};
+        // before interp range, do not fill
+        if (start.key == INT64_MIN || end.key == INT64_MAX) {
+          break;
+        }
 
-              int32_t srcType = pCtx->inputType;
-              if (isNull((char *)&pCtx->start.val, srcType) || isNull((char *)&pCtx->end.val, srcType)) {
-                setNull(pCtx->pOutput, srcType, pCtx->inputBytes);
-              } else {
-                bool exceedMax = false, exceedMin = false;
-                taosGetLinearInterpolationVal(&point, pCtx->outputType, &point1, &point2, TSDB_DATA_TYPE_DOUBLE, &exceedMax, &exceedMin);
-                if (exceedMax || exceedMin) {
-                  __compar_fn_t func = getComparFunc((int32_t)pCtx->inputType, 0);
-                  if (func(&pCtx->start.val, &pCtx->end.val) <= 0) {
-                    COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, exceedMax ? &pCtx->start.val : &pCtx->end.val);
-                  } else {
-                    COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, exceedMax ? &pCtx->end.val : &pCtx->start.val);
-                  }
-                }
-              }
-#endif
-        // TODO: pResBlock->info.rows += 1;
+        if (pLinearInfo->hasNull) {
+          colDataAppendNULL(pDst, rows);
+        } else {
+          taosGetLinearInterpolationVal(&current, pLinearInfo->type, &start, &end, pLinearInfo->type);
+          colDataAppend(pDst, rows, (char *)current.val, false);
+        }
+
+        pResBlock->info.rows += 1;
         break;
       }
       case TSDB_FILL_PREV: {
@@ -2246,6 +2263,55 @@ static int32_t initNextRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pB
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t initFillLinearInfo(STimeSliceOperatorInfo* pInfo, SSDataBlock* pBlock) {
+  if (pInfo->pLinearInfo != NULL) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  pInfo->pLinearInfo = taosArrayInit(4, sizeof(SFillLinearInfo));
+  if (pInfo->pNextRow == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
+
+    SFillLinearInfo linearInfo = {0};
+    linearInfo.start.key = INT64_MIN;
+    linearInfo.end.key   = INT64_MAX;
+    linearInfo.start.val = taosMemoryCalloc(1, pColInfo->info.bytes);
+    linearInfo.end.val   = taosMemoryCalloc(1, pColInfo->info.bytes);
+    linearInfo.hasNull   = false;
+    linearInfo.fillLastPoint = false;
+    linearInfo.type  = pColInfo->info.type;
+    linearInfo.bytes = pColInfo->info.bytes;
+    taosArrayPush(pInfo->pLinearInfo, &linearInfo);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t initKeeperInfo(STimeSliceOperatorInfo* pInfo, SSDataBlock* pBlock) {
+  int32_t code;
+  code = initPrevRowsKeeper(pInfo, pBlock);
+  if (code != TSDB_CODE_SUCCESS) {
+    return TSDB_CODE_FAILED;
+  }
+
+  code = initNextRowsKeeper(pInfo, pBlock);
+  if (code != TSDB_CODE_SUCCESS) {
+    return TSDB_CODE_FAILED;
+  }
+
+  code = initFillLinearInfo(pInfo, pBlock);
+  if (code != TSDB_CODE_SUCCESS) {
+    return TSDB_CODE_FAILED;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
   if (pOperator->status == OP_EXEC_DONE) {
     return NULL;
@@ -2278,13 +2344,7 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
       break;
     }
 
-    int32_t code;
-    code = initPrevRowsKeeper(pSliceInfo, pBlock);
-    if (code != TSDB_CODE_SUCCESS) {
-      longjmp(pTaskInfo->env, code);
-    }
-
-    code = initNextRowsKeeper(pSliceInfo, pBlock);
+    int32_t code = initKeeperInfo(pSliceInfo, pBlock);
     if (code != TSDB_CODE_SUCCESS) {
       longjmp(pTaskInfo->env, code);
     }
@@ -2312,46 +2372,103 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
         pResBlock->info.rows += 1;
         doKeepPrevRows(pSliceInfo, pBlock, i);
 
-        pSliceInfo->current =
-            taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
-        if (pSliceInfo->current > pSliceInfo->win.ekey) {
-          doSetOperatorCompleted(pOperator);
-          break;
-        }
+        // for linear interpolation, always fill value between this and next points;
+        // if its the first point in data block, also fill values between previous(if there's any) and this point;
+        // if its the last point in data block, no need to fill, but reserve this point as the start value for next data block.
+        if (pSliceInfo->fillType == TSDB_FILL_LINEAR) {
+          doKeepLinearInfo(pSliceInfo, pBlock, i);
+          pSliceInfo->current =
+              taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+          if (i < pBlock->info.rows - 1) {
+            int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
+            if (nextTs > pSliceInfo->current) {
+              while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
+                genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
+                pSliceInfo->current =
+                    taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+                if (pResBlock->info.rows >= pResBlock->info.capacity) {
+                  break;
+                }
+              }
 
-        if (pResBlock->info.rows >= pResBlock->info.capacity) {
-          break;
+              if (pSliceInfo->current > pSliceInfo->win.ekey) {
+                doSetOperatorCompleted(pOperator);
+                break;
+              }
+            } else {
+              // ignore current row, and do nothing
+            }
+          } else { // it is the last row of current block
+          }
+        } else { // non-linear interpolation
+          pSliceInfo->current =
+              taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+          if (pSliceInfo->current > pSliceInfo->win.ekey) {
+            doSetOperatorCompleted(pOperator);
+            break;
+          }
+
+          if (pResBlock->info.rows >= pResBlock->info.capacity) {
+            break;
+          }
         }
       } else if (ts < pSliceInfo->current) {
-        // in case interpolation window starts and ends between two datapoints, fill(prev) need to interpolate
+        // in case of interpolation window starts and ends between two datapoints, fill(prev) need to interpolate
         doKeepPrevRows(pSliceInfo, pBlock, i);
 
-        if (i < pBlock->info.rows - 1) {
-          // in case interpolation window starts and ends between two datapoints, fill(next) need to interpolate
-          doKeepNextRows(pSliceInfo, pBlock, i + 1);
-          int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
-          if (nextTs > pSliceInfo->current) {
-            while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
-              genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
-              pSliceInfo->current =
-                  taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
-              if (pResBlock->info.rows >= pResBlock->info.capacity) {
+        if (pSliceInfo->fillType == TSDB_FILL_LINEAR) {
+          doKeepLinearInfo(pSliceInfo, pBlock, i);
+          //pSliceInfo->current =
+          //    taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+          if (i < pBlock->info.rows - 1) {
+            int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
+            if (nextTs > pSliceInfo->current) {
+              while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
+                genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
+                pSliceInfo->current =
+                    taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+                if (pResBlock->info.rows >= pResBlock->info.capacity) {
+                  break;
+                }
+              }
+
+              if (pSliceInfo->current > pSliceInfo->win.ekey) {
+                doSetOperatorCompleted(pOperator);
                 break;
               }
+            } else {
+              // ignore current row, and do nothing
             }
+          } else { // it is the last row of current block
+          }
+        } else { // non-linear interpolation
+          if (i < pBlock->info.rows - 1) {
+            // in case of interpolation window starts and ends between two datapoints, fill(next) need to interpolate
+            doKeepNextRows(pSliceInfo, pBlock, i + 1);
+            int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
+            if (nextTs > pSliceInfo->current) {
+              while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
+                genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
+                pSliceInfo->current =
+                    taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+                if (pResBlock->info.rows >= pResBlock->info.capacity) {
+                  break;
+                }
+              }
 
-            if (pSliceInfo->current > pSliceInfo->win.ekey) {
-              doSetOperatorCompleted(pOperator);
-              break;
+              if (pSliceInfo->current > pSliceInfo->win.ekey) {
+                doSetOperatorCompleted(pOperator);
+                break;
+              }
+            } else {
+              // ignore current row, and do nothing
             }
-          } else {
-            // ignore current row, and do nothing
+          } else {  // it is the last row of current block
+            doKeepPrevRows(pSliceInfo, pBlock, i);
           }
-        } else {  // it is the last row of current block
-          doKeepPrevRows(pSliceInfo, pBlock, i);
         }
       } else {  // ts > pSliceInfo->current
-        // in case interpolation window starts and ends between two datapoints, fill(next) need to interpolate
+        // in case of interpolation window starts and ends between two datapoints, fill(next) need to interpolate
         doKeepNextRows(pSliceInfo, pBlock, i);
 
         while (pSliceInfo->current < ts && pSliceInfo->current <= pSliceInfo->win.ekey) {
@@ -2380,11 +2497,39 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
           pResBlock->info.rows += 1;
           doKeepPrevRows(pSliceInfo, pBlock, i);
 
-          pSliceInfo->current =
-              taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
 
-          if (pResBlock->info.rows >= pResBlock->info.capacity) {
-            break;
+          if (pSliceInfo->fillType == TSDB_FILL_LINEAR) {
+            doKeepLinearInfo(pSliceInfo, pBlock, i);
+            pSliceInfo->current =
+                taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+            if (i < pBlock->info.rows - 1) {
+              int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
+              if (nextTs > pSliceInfo->current) {
+                while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
+                  genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
+                  pSliceInfo->current =
+                      taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+                  if (pResBlock->info.rows >= pResBlock->info.capacity) {
+                    break;
+                  }
+                }
+
+                if (pSliceInfo->current > pSliceInfo->win.ekey) {
+                  doSetOperatorCompleted(pOperator);
+                  break;
+                }
+              } else {
+                // ignore current row, and do nothing
+              }
+            } else { // it is the last row of current block
+            }
+          } else { // non-linear interpolation
+            pSliceInfo->current =
+                taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+
+            if (pResBlock->info.rows >= pResBlock->info.capacity) {
+              break;
+            }
           }
         }
 
@@ -2446,6 +2591,9 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode
   pInfo->fillType = convertFillType(pInterpPhyNode->fillMode);
   initResultSizeInfo(&pOperator->resultInfo, 4096);
 
+  pInfo->pPrevRow = NULL;
+  pInfo->pNextRow = NULL;
+  pInfo->pLinearInfo = NULL;
   pInfo->pFillColInfo = createFillColInfo(pExprInfo, numOfExprs, (SNodeListNode*)pInterpPhyNode->pFillValues);
   pInfo->pRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
   pInfo->win = pInterpPhyNode->timeRange;