[TD-28]fix rate function bugs in reversed traverse query, and interpolation...

[TD-28]fix rate function bugs in reversed traverse query, and interpolation for start and end timestamp of query time window

src/client/inc/tsclient.h

@@ -197,7 +197,7 @@ typedef struct SDataBlockList {
 typedef struct SQueryInfo {
   int16_t  command;  // the command may be different for each subclause, so keep it seperately.
   uint16_t type;     // query/insert/import type
-  char     intervalTimeUnit;
+  char     slidingTimeUnit;
 
   int64_t         etime, stime;
   int64_t         intervalTime;  // aggregation time interval

src/client/src/tscFunctionImpl.c

@@ -4400,7 +4400,7 @@ static double do_calc_rate(const SRateInfo* pRateInfo) {
     }
   }
 
-  double duration = (pRateInfo->lastKey - pRateInfo->firstKey) / 1000;
+  double duration = (pRateInfo->lastKey - pRateInfo->firstKey) / 1000.0;
   double resultVal = diff / duration;
 
   pTrace("do_calc_rate() isIRate:%d firstKey:%" PRId64 " lastKey:%" PRId64 " firstValue:%f lastValue:%f CorrectionValue:%f resultVal:%f", 

src/client/src/tscSQLParser.c

@@ -598,9 +598,6 @@ int32_t parseIntervalClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
     pQueryInfo->intervalTime = pQueryInfo->intervalTime / 1000;
   }
 
-  /* parser has filter the illegal type, no need to check here */
-  pQueryInfo->intervalTimeUnit = pQuerySql->interval.z[pQuerySql->interval.n - 1];
-
   // interval cannot be less than 10 milliseconds
   if (pQueryInfo->intervalTime < tsMinIntervalTime) {
     return invalidSqlErrMsg(pQueryInfo->msg, msg2);
@@ -689,10 +686,15 @@ int32_t parseSlidingClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
     if (pQueryInfo->slidingTime > pQueryInfo->intervalTime) {
       return invalidSqlErrMsg(pQueryInfo->msg, msg1);
     }
+  
+    pQueryInfo->slidingTimeUnit = pQuerySql->sliding.z[pQuerySql->sliding.n - 1];
   } else {
     pQueryInfo->slidingTime = pQueryInfo->intervalTime;
+  
+    // parser has filter the illegal type, no need to check here
+    pQueryInfo->slidingTimeUnit = pQuerySql->interval.z[pQuerySql->interval.n - 1];
   }
-
+  
   return TSDB_CODE_SUCCESS;
 }
 

src/client/src/tscSecondaryMerge.c

@@ -324,7 +324,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
 
   int64_t stime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->stime : pQueryInfo->etime;
   int64_t revisedSTime =
-      taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->intervalTimeUnit, prec);
+      taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, prec);
 
   SInterpolationInfo *pInterpoInfo = &pReducer->interpolationInfo;
   taosInitInterpoInfo(pInterpoInfo, pQueryInfo->order.order, revisedSTime, pQueryInfo->groupbyExpr.numOfGroupCols,
@@ -779,7 +779,7 @@ void savePrevRecordAndSetupInterpoInfo(SLocalReducer *pLocalReducer, SQueryInfo
 
   int64_t stime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->stime : pQueryInfo->etime;
   int64_t revisedSTime =
-      taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->intervalTimeUnit, prec);
+      taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, prec);
 
   taosInitInterpoInfo(pInterpoInfo, pQueryInfo->order.order, revisedSTime, pQueryInfo->groupbyExpr.numOfGroupCols,
                       pLocalReducer->rowSize);
@@ -923,7 +923,7 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
   while (1) {
     int32_t remains = taosNumOfRemainPoints(pInterpoInfo);
     TSKEY   etime = taosGetRevisedEndKey(actualETime, pQueryInfo->order.order, pQueryInfo->intervalTime,
-                                       pQueryInfo->intervalTimeUnit, precision);
+                                       pQueryInfo->slidingTimeUnit, precision);
     int32_t nrows = taosGetNumOfResultWithInterpo(pInterpoInfo, pPrimaryKeys, remains, pQueryInfo->intervalTime, etime,
                                                   pLocalReducer->resColModel->capacity);
 
@@ -1275,7 +1275,7 @@ static void resetEnvForNewResultset(SSqlRes *pRes, SSqlCmd *pCmd, SLocalReducer
   if (pQueryInfo->interpoType != TSDB_INTERPO_NONE) {
     int64_t stime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->stime : pQueryInfo->etime;
     int64_t newTime =
-        taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->intervalTimeUnit, precision);
+        taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, precision);
 
     taosInitInterpoInfo(&pLocalReducer->interpolationInfo, pQueryInfo->order.order, newTime,
                         pQueryInfo->groupbyExpr.numOfGroupCols, pLocalReducer->rowSize);
@@ -1305,7 +1305,7 @@ static bool doInterpolationForCurrentGroup(SSqlObj *pSql) {
 
     int32_t remain = taosNumOfRemainPoints(pInterpoInfo);
     TSKEY   ekey =
-        taosGetRevisedEndKey(etime, pQueryInfo->order.order, pQueryInfo->intervalTime, pQueryInfo->intervalTimeUnit, p);
+        taosGetRevisedEndKey(etime, pQueryInfo->order.order, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, p);
     int32_t rows = taosGetNumOfResultWithInterpo(pInterpoInfo, (TSKEY *)pLocalReducer->pBufForInterpo, remain,
                                                  pQueryInfo->intervalTime, ekey, pLocalReducer->resColModel->capacity);
     if (rows > 0) {  // do interpo
@@ -1338,7 +1338,7 @@ static bool doHandleLastRemainData(SSqlObj *pSql) {
       int64_t etime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->etime : pQueryInfo->stime;
 
       etime = taosGetRevisedEndKey(etime, pQueryInfo->order.order, pQueryInfo->intervalTime,
-                                   pQueryInfo->intervalTimeUnit, precision);
+                                   pQueryInfo->slidingTimeUnit, precision);
       int32_t rows = taosGetNumOfResultWithInterpo(pInterpoInfo, NULL, 0, pQueryInfo->intervalTime, etime,
                                                    pLocalReducer->resColModel->capacity);
       if (rows > 0) {  // do interpo

src/client/src/tscServer.c

@@ -1680,7 +1680,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
   }
 
   pQueryMsg->intervalTime = htobe64(pQueryInfo->intervalTime);
-  pQueryMsg->intervalTimeUnit = pQueryInfo->intervalTimeUnit;
+  pQueryMsg->slidingTimeUnit = pQueryInfo->slidingTimeUnit;
   pQueryMsg->slidingTime = htobe64(pQueryInfo->slidingTime);
   
   if (pQueryInfo->intervalTime < 0) {

src/inc/taosmsg.h

@@ -513,7 +513,7 @@ typedef struct {
   int16_t orderColId;
 
   int16_t numOfCols;         // the number of columns will be load from vnode
-  char    intervalTimeUnit;  // time interval type, for revisement of interval(1d)
+  char    slidingTimeUnit;   // time interval type, for revisement of interval(1d)
 
   int64_t intervalTime;  // time interval for aggregation, in million second
   int64_t slidingTime;       // value for sliding window

src/system/detail/inc/vnode.h

@@ -261,7 +261,7 @@ typedef struct SQuery {
   TSKEY       ekey;
   int64_t     intervalTime;
   int64_t     slidingTime;       // sliding time for sliding window query
-  char        intervalTimeUnit;  // interval data type, used for daytime revise
+  char        slidingTimeUnit;  // interval data type, used for daytime revise
   int8_t      precision;
   int16_t     numOfOutputCols;
   int16_t     interpoType;

src/system/detail/src/vnodeQueryImpl.c

@@ -1663,8 +1663,12 @@ static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey,
         continue;
       }
 
-      if ((pResult->window.ekey <= lastKey && QUERY_IS_ASC_QUERY(pQuery)) ||
-          (pResult->window.skey >= lastKey && !QUERY_IS_ASC_QUERY(pQuery))) {
+      /*
+       * when the ekey equals to lastKey of current block, do NOT close it, since the interpolation may
+       * be involved.
+       */
+      if ((pResult->window.ekey < lastKey && QUERY_IS_ASC_QUERY(pQuery)) ||
+          (pResult->window.skey > lastKey && !QUERY_IS_ASC_QUERY(pQuery))) {
         closeTimeWindow(pWindowResInfo, i);
       } else {
         skey = pResult->window.skey;
@@ -1998,7 +2002,7 @@ static void interpolateStartKeyValue(SQueryRuntimeEnv *pRuntimeEnv, SBlockInfo*
   
   if (QUERY_IS_ASC_QUERY(pQuery)) {
     // the queried time window and time window of data block must be intersect
-    assert(win->ekey >= pBlockInfo->keyFirst && win->skey <= pBlockInfo->keyLast);
+//    assert(win->ekey >= pBlockInfo->keyFirst && win->skey <= pBlockInfo->keyLast);
     
     /*
      * this win should not be the first time window that starts from a less timestamp than
@@ -2151,20 +2155,27 @@ static int32_t blockwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, int32_t
     // get current not closed time window
     int32_t slot = pWindowResInfo->curIndex;
     if (slot != -1) {
-      STimeWindow w = getWindowResult(pWindowResInfo, slot)->window;
-  
-      // if current window is closed and locates before current active block, interpolate the result and close it
-      if (w.skey != win.skey || w.ekey != win.ekey) {
-        assert((w.ekey < win.skey && w.ekey < ts && QUERY_IS_ASC_QUERY(pQuery)) ||
-            (w.skey > win.ekey && w.skey > ts && !QUERY_IS_ASC_QUERY(pQuery)));
-    
-        forwardStep = 0;
-        doSetInterpolationDataForTimeWindow(pRuntimeEnv, pWindowResInfo, pBlockInfo, &w, offset, forwardStep);
-    
-        SWindowStatus *pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo));
-        doBlockwiseApplyFunctions(pRuntimeEnv, pStatus, &w, pQuery->pos, forwardStep);
-    
-        closeTimeWindow(pWindowResInfo, curTimeWindow(pWindowResInfo));
+      while (slot < pWindowResInfo->size) {
+        STimeWindow w = getWindowResult(pWindowResInfo, slot)->window;
+
+        // if current active window locates before current data block, do interpolate the result and close it
+        if (w.skey != win.skey || w.ekey != win.ekey) {
+          assert((w.skey < win.skey && w.ekey < ts && QUERY_IS_ASC_QUERY(pQuery)) ||
+                 (w.skey > win.skey && w.skey > ts && !QUERY_IS_ASC_QUERY(pQuery)));
+
+          forwardStep = 0;
+          doSetInterpolationDataForTimeWindow(pRuntimeEnv, pWindowResInfo, pBlockInfo, &w, offset, forwardStep);
+
+          SWindowStatus *pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo));
+          doBlockwiseApplyFunctions(pRuntimeEnv, pStatus, &w, pQuery->pos, forwardStep);
+
+          closeTimeWindow(pWindowResInfo, curTimeWindow(pWindowResInfo));
+        } else {
+          break;
+        }
+
+        // try next time window
+        slot += 1;
       }
     }
     
@@ -2441,6 +2452,26 @@ void closeAllTimeWindow(SWindowResInfo *pWindowResInfo) {
   }
 }
 
+/*
+ * remove the results that are not the FIRST time window that spreads beyond the
+ * the last qualified time stamp in case of sliding query, which the sliding time is not equalled to the interval time
+ */
+void removeRedundantWindow(SWindowResInfo *pWindowResInfo, TSKEY lastKey, int32_t order) {
+  assert(pWindowResInfo->size >= 0 && pWindowResInfo->capacity >= pWindowResInfo->size);
+  
+  int32_t i = 0;
+  while(i < pWindowResInfo->size &&
+      ((pWindowResInfo->pResult[i].window.ekey < lastKey && order == QUERY_ASC_FORWARD_STEP) ||
+          (pWindowResInfo->pResult[i].window.skey > lastKey && order == QUERY_DESC_FORWARD_STEP))) {
+    ++i;
+  }
+
+//  assert(i < pWindowResInfo->size);
+  if (i < pWindowResInfo->size) {
+    pWindowResInfo->size = (i + 1);
+  }
+}
+
 static int32_t setGroupResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, char *pData, int16_t type, int16_t bytes) {
   if (isNull(pData, type)) {  // ignore the null value
     return -1;
@@ -3614,7 +3645,7 @@ void vnodeCheckIfDataExists(SQueryRuntimeEnv *pRuntimeEnv, SMeterObj *pMeterObj,
 void doGetAlignedIntervalQueryRangeImpl(SQuery *pQuery, int64_t pKey, int64_t keyFirst, int64_t keyLast,
                                         int64_t *actualSkey, int64_t *actualEkey, int64_t *skey, int64_t *ekey) {
   assert(pKey >= keyFirst && pKey <= keyLast);
-  *skey = taosGetIntervalStartTimestamp(pKey, pQuery->intervalTime, pQuery->intervalTimeUnit, pQuery->precision);
+  *skey = taosGetIntervalStartTimestamp(pKey, pQuery->slidingTime, pQuery->slidingTimeUnit, pQuery->precision);
 
   if (keyFirst > (INT64_MAX - pQuery->intervalTime)) {
     /*
@@ -5065,7 +5096,7 @@ int32_t vnodeQueryTablePrepare(SQInfo *pQInfo, SMeterObj *pMeterObj, STableQuery
     return TSDB_CODE_SUCCESS;
   }
 
-  int64_t rs = taosGetIntervalStartTimestamp(pSupporter->rawSKey, pQuery->intervalTime, pQuery->intervalTimeUnit,
+  int64_t rs = taosGetIntervalStartTimestamp(pSupporter->rawSKey, pQuery->intervalTime, pQuery->slidingTimeUnit,
                                              pQuery->precision);
   taosInitInterpoInfo(&pRuntimeEnv->interpoInfo, pQuery->order.order, rs, 0, 0);
   allocMemForInterpo(pSupporter, pQuery, pMeterObj);
@@ -5201,7 +5232,7 @@ int32_t vnodeSTableQueryPrepare(SQInfo *pQInfo, SQuery *pQuery, void *param) {
   }
 
   TSKEY revisedStime = taosGetIntervalStartTimestamp(pSupporter->rawSKey, pQuery->intervalTime,
-                                                     pQuery->intervalTimeUnit, pQuery->precision);
+                                                     pQuery->slidingTimeUnit, pQuery->precision);
   taosInitInterpoInfo(&pRuntimeEnv->interpoInfo, pQuery->order.order, revisedStime, 0, 0);
   pRuntimeEnv->stableQuery = true;
 
@@ -5647,6 +5678,8 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
   if (isIntervalQuery(pQuery) && IS_MASTER_SCAN(pRuntimeEnv)) {
     if (Q_STATUS_EQUAL(pQuery->over, QUERY_COMPLETED | QUERY_NO_DATA_TO_CHECK)) {
       closeAllTimeWindow(&pRuntimeEnv->windowResInfo);
+      removeRedundantWindow(&pRuntimeEnv->windowResInfo, pQuery->lastKey - step, step);
+  
       pRuntimeEnv->windowResInfo.curIndex = pRuntimeEnv->windowResInfo.size - 1;
     } else if (Q_STATUS_EQUAL(pQuery->over, QUERY_RESBUF_FULL)) {  // check if window needs to be closed
       SBlockInfo blockInfo = getBlockInfo(pRuntimeEnv);
@@ -7789,7 +7822,7 @@ bool vnodeHasRemainResults(void *handle) {
     // query has completed
     if (Q_STATUS_EQUAL(pQuery->over, QUERY_COMPLETED | QUERY_NO_DATA_TO_CHECK)) {
       TSKEY   ekey = taosGetRevisedEndKey(pSupporter->rawEKey, pQuery->order.order, pQuery->intervalTime,
-                                        pQuery->intervalTimeUnit, pQuery->precision);
+                                        pQuery->slidingTimeUnit, pQuery->precision);
       int32_t numOfTotal = taosGetNumOfResultWithInterpo(pInterpoInfo, (TSKEY *)pRuntimeEnv->pInterpoBuf[0]->data,
                                                          remain, pQuery->intervalTime, ekey, pQuery->pointsToRead);
       return numOfTotal > 0;
@@ -7900,7 +7933,7 @@ int32_t vnodeQueryResultInterpolate(SQInfo *pQInfo, tFilePage **pDst, tFilePage
     numOfRows = taosNumOfRemainPoints(&pRuntimeEnv->interpoInfo);
 
     TSKEY   ekey = taosGetRevisedEndKey(pSupporter->rawEKey, pQuery->order.order, pQuery->intervalTime,
-                                      pQuery->intervalTimeUnit, pQuery->precision);
+                                      pQuery->slidingTimeUnit, pQuery->precision);
     int32_t numOfFinalRows = taosGetNumOfResultWithInterpo(&pRuntimeEnv->interpoInfo, (TSKEY *)pDataSrc[0]->data,
                                                            numOfRows, pQuery->intervalTime, ekey, pQuery->pointsToRead);
 

src/system/detail/src/vnodeRead.c

@@ -269,7 +269,7 @@ static SQInfo *vnodeAllocateQInfoEx(SQueryMeterMsg *pQueryMsg, SSqlGroupbyExpr *
   pQuery->intervalTime = pQueryMsg->intervalTime;
   pQuery->slidingTime = pQueryMsg->slidingTime;
   pQuery->interpoType = pQueryMsg->interpoType;
-  pQuery->intervalTimeUnit = pQueryMsg->intervalTimeUnit;
+  pQuery->slidingTimeUnit = pQueryMsg->slidingTimeUnit;
 
   pQInfo->query.pointsToRead = vnodeList[pMeterObj->vnode].cfg.rowsInFileBlock;
 

src/util/src/tinterpolation.c

@@ -22,12 +22,12 @@
 
 #define INTERPOL_IS_ASC_INTERPOL(interp) ((interp)->order == TSQL_SO_ASC)
 
-int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t timeRange, char intervalTimeUnit, int16_t precision) {
+int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t timeRange, char slidingTimeUnit, int16_t precision) {
   if (timeRange == 0) {
     return startTime;
   }
 
-  if (intervalTimeUnit == 'a' || intervalTimeUnit == 'm' || intervalTimeUnit == 's' || intervalTimeUnit == 'h') {
+  if (slidingTimeUnit == 'a' || slidingTimeUnit == 'm' || slidingTimeUnit == 's' || slidingTimeUnit == 'h') {
     return (startTime / timeRange) * timeRange;
   } else {
     /*
@@ -95,11 +95,11 @@ void taosInterpoSetStartInfo(SInterpolationInfo* pInterpoInfo, int32_t numOfRawD
   pInterpoInfo->numOfRawDataInRows = numOfRawDataInRows;
 }
 
-TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int32_t timeInterval, int8_t intervalTimeUnit, int8_t precision) {
+TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int32_t timeInterval, int8_t slidingTimeUnit, int8_t precision) {
   if (order == TSQL_SO_ASC) {
     return ekey;
   } else {
-    return taosGetIntervalStartTimestamp(ekey, timeInterval, intervalTimeUnit, precision);
+    return taosGetIntervalStartTimestamp(ekey, timeInterval, slidingTimeUnit, precision);
   }
 }