From 3d931faa5bc3c530aa65e9ffc10a9f8496c53afb Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 1 May 2022 00:23:22 +0800
Subject: [PATCH] refactor(query): do some internal refactor.

---
 source/libs/function/src/builtinsimpl.c | 106 ++++++++++++++++++------
 1 file changed, 82 insertions(+), 24 deletions(-)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index bb01af420d..05dddb50aa 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -78,6 +78,8 @@ typedef struct SDiffInfo {
     int64_t i64;
     double  d64;
   } prev;
+
+  int64_t prevTs;
 } SDiffInfo;
 
 typedef struct SSpreadInfo {
@@ -1196,8 +1198,8 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
   bool    isFirstBlock = (pDiffInfo->hasPrev == false);
   int32_t numOfElems = 0;
 
-  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pCtx->order);
-  //  int32_t i = (pCtx->order == TSDB_ORDER_ASC) ? 0 : pCtx->size - 1;
+//  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pCtx->order);
+//  int32_t start = (pCtx->order == TSDB_ORDER_ASC) ? pInput->startRowIndex : pInput->numOfRows + pInput->startRowIndex - 1;
 
   SColumnInfoData* pTsOutput = pCtx->pTsOutput;
   TSKEY*           tsList = (int64_t*)pInput->pPTS->pData;
@@ -1206,44 +1208,100 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
   switch (pInputCol->info.type) {
     case TSDB_DATA_TYPE_INT: {
       SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
-      for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += step) {
-        int32_t pos = startOffset + (isFirstBlock ? (numOfElems - 1) : numOfElems);
-        if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
-          if (pDiffInfo->includeNull) {
-            colDataSetNull_f(pOutput->nullbitmap, pos);
-            if (tsList != NULL) {
-              colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+      if (pCtx->order == TSDB_ORDER_ASC) {
+        for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
+          int32_t pos = startOffset + (isFirstBlock ? (numOfElems - 1) : numOfElems);
+          if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
+            if (pDiffInfo->includeNull) {
+              colDataSetNull_f(pOutput->nullbitmap, pos);
+              if (tsList != NULL) {
+                colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+              }
+
+              numOfElems += 1;
+            }
+            continue;
+          }
+
+          int32_t v = *(int32_t*)colDataGetData(pInputCol, i);
+          if (pDiffInfo->hasPrev) {
+            int32_t delta = (int32_t)(v - pDiffInfo->prev.i64);  // direct previous may be null
+            if (delta < 0 && pDiffInfo->ignoreNegative) {
+              colDataSetNull_f(pOutput->nullbitmap, pos);
+            } else {
+              colDataAppendInt32(pOutput, pos, &delta);
             }
 
-            numOfElems += 1;
+            if (pTsOutput != NULL) {
+              colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+            }
           }
-          continue;
+
+          pDiffInfo->prev.i64 = v;
+          pDiffInfo->hasPrev = true;
+          numOfElems++;
         }
+      } else {
+        for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
+          int32_t v = *(int32_t*)colDataGetData(pInputCol, i);
+          int32_t pos = startOffset + numOfElems;
+
+          if (pDiffInfo->hasPrev) {
+            int32_t delta = -(int32_t)(v - pDiffInfo->prev.i64);  // direct previous may be null
+            if (delta < 0 && pDiffInfo->ignoreNegative) {
+              colDataSetNull_f(pOutput->nullbitmap, pos);
+            } else {
+              colDataAppendInt32(pOutput, pos, &delta);
+            }
 
-        int32_t v = *(int32_t*)colDataGetData(pInputCol, i);
-        if (pDiffInfo->hasPrev) {
-          int32_t delta = (int32_t)(v - pDiffInfo->prev.i64);  // direct previous may be null
-          if (delta < 0 && pDiffInfo->ignoreNegative) {
-            colDataSetNull_f(pOutput->nullbitmap, pos);
-          } else {
-            colDataAppendInt32(pOutput, pos, &delta);
+            if (pTsOutput != NULL) {
+              colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+            }
+            pDiffInfo->hasPrev = false;
           }
 
-          if (pTsOutput != NULL) {
-            colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+          if (i < pInput->numOfRows + pInput->startRowIndex - 1) {
+            int32_t next = *(int32_t*)colDataGetData(pInputCol, i + 1);
+
+            int32_t delta = v - next;  // direct previous may be null
+            colDataAppendInt32(pOutput, pos, &delta);
+
+            if (pTsOutput != NULL) {
+              colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+            }
+          } else {
+            pDiffInfo->prev.i64 = v;
+            if (pTsOutput != NULL) {
+              pDiffInfo->prevTs = tsList[i];
+            }
+            pDiffInfo->hasPrev = true;
           }
+//          if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
+//            if (pDiffInfo->includeNull) {
+//              colDataSetNull_f(pOutput->nullbitmap, pos);
+//              if (tsList != NULL) {
+//                colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+//              }
+//
+//              numOfElems += 1;
+//            }
+//            continue;
+//          }
+
+//          int32_t v = *(int32_t*)colDataGetData(pInputCol, i);
+
+//          pDiffInfo->prev.i64 = v;
+//          pDiffInfo->hasPrev = true;
+          numOfElems++;
         }
 
-        pDiffInfo->prev.i64 = v;
-        pDiffInfo->hasPrev = true;
-        numOfElems++;
       }
       break;
     }
 
     case TSDB_DATA_TYPE_BIGINT: {
       SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
-      for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += step) {
+      for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
         if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
           continue;
         }
-- 
GitLab