enh(query): add derivative function.

include/libs/function/functionMgt.h

@@ -134,6 +134,8 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_HYPERLOGLOG_MERGE,
   FUNCTION_TYPE_ELAPSED_PARTIAL,
   FUNCTION_TYPE_ELAPSED_MERGE,
+  FUNCTION_TYPE_SAMPLE_PARTIAL,
+  FUNCTION_TYPE_SAMPLE_MERGE,
 
   // user defined funcion
   FUNCTION_TYPE_UDF = 10000

source/libs/function/inc/builtinsimpl.h

@@ -91,6 +91,10 @@ bool getDiffFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool diffFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo);
 int32_t diffFunction(SqlFunctionCtx *pCtx);
 
+bool getDerivativeFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+bool derivativeFuncSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo);
+int32_t derivativeFunction(SqlFunctionCtx *pCtx);
+
 bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 int32_t firstFunction(SqlFunctionCtx *pCtx);
 int32_t lastFunction(SqlFunctionCtx *pCtx);

source/libs/function/inc/texpr.h

@@ -60,12 +60,6 @@ typedef struct SExprTraverseSupp {
   void                  *pExtInfo;
 } SExprTraverseSupp;
 
-tExprNode* exprTreeFromTableName(const char* tbnameCond);
-
-bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp *param);
-
-void buildFilterSetFromBinary(void **q, const char *buf, int32_t len);
-
 #ifdef __cplusplus
 }
 #endif

source/libs/function/inc/tunaryoperator.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_COMMON_UNARY_SCALAR_OPERATOR_H_
-#define _TD_COMMON_UNARY_SCALAR_OPERATOR_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-//#include "tscalarfunction.h"
-
-//typedef void (*_unary_scalar_fn_t)(SScalarParam *pLeft, SScalarParam* pOutput);
-//_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t binOperator);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif  /*_TD_COMMON_BIN_SCALAR_OPERATOR_H_*/

source/libs/function/src/builtins.c

@@ -884,6 +884,38 @@ static int32_t translateLastRow(SFunctionNode* pFunc, char* pErrBuf, int32_t len
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateDerivative(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  if (3 != LIST_LENGTH(pFunc->pParameterList)) {
+    return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
+
+  // param1
+  SNode* pParamNode1 = nodesListGetNode(pFunc->pParameterList, 1);
+  if (QUERY_NODE_VALUE != nodeType(pParamNode1)) {
+    return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  SValueNode* pValue = (SValueNode*)pParamNode1;
+  pValue->notReserved = true;
+
+  if (!IS_NUMERIC_TYPE(colType)) {
+    return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  SNode* pParamNode2 = nodesListGetNode(pFunc->pParameterList, 2);
+  SValueNode* pValue2 = (SValueNode*)pParamNode2;
+  pValue2->notReserved = true;
+
+  if (pValue2->datum.i != 0 && pValue2->datum.i != 1) {
+    return invaildFuncParaValueErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes, .type = TSDB_DATA_TYPE_DOUBLE};
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   // first(col_list) will be rewritten as first(col)
   if (1 != LIST_LENGTH(pFunc->pParameterList)) {
@@ -1583,6 +1615,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .invertFunc   = NULL,
     .combineFunc  = elapsedCombine,
   },
+  {
+    .name = "derivative",
+    .type = FUNCTION_TYPE_DERIVATIVE,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .translateFunc = translateDerivative,
+    .getEnvFunc   = getDerivativeFuncEnv,
+    .initFunc     = derivativeFuncSetup,
+    .processFunc  = derivativeFunction,
+    .finalizeFunc = functionFinalize
+  },
   {
     .name = "last_row",
     .type = FUNCTION_TYPE_LAST_ROW,

source/libs/function/src/builtinsimpl.c

@@ -4879,3 +4879,81 @@ int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 
   return row;
 }
+
+typedef struct SDerivInfo {
+  double   prevValue;     // previous value
+  TSKEY    prevTs;        // previous timestamp
+  bool     ignoreNegative;// ignore the negative value
+  int64_t  tsWindow;      // time window for derivative
+  bool     valueSet;      // the value has been set already
+} SDerivInfo;
+
+bool getDerivativeFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(SDerivInfo);
+  return true;
+}
+
+bool derivativeFuncSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo) {
+  if (!functionSetup(pCtx, pResInfo)) {
+    return false;  // not initialized since it has been initialized
+  }
+
+  SDerivInfo* pDerivInfo = GET_ROWCELL_INTERBUF(pResInfo);
+
+  pDerivInfo->ignoreNegative = pCtx->param[2].param.i;
+  pDerivInfo->prevTs   = -1;
+  pDerivInfo->tsWindow = pCtx->param[1].param.i;
+  pDerivInfo->valueSet = false;
+  return true;
+}
+
+int32_t derivativeFunction(SqlFunctionCtx *pCtx) {
+  SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
+  SDerivInfo* pDerivInfo = GET_ROWCELL_INTERBUF(pResInfo);
+
+  SInputColumnInfoData* pInput = &pCtx->input;
+  SColumnInfoData* pInputCol = pInput->pData[0];
+  
+  int32_t numOfElems = 0;
+  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
+  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
+
+  int32_t step = 1;
+  int32_t i = pInput->startRowIndex;
+  TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
+
+  if (pCtx->order == TSDB_ORDER_ASC) {
+    double v = 0;
+
+    for (; i < pInput->numOfRows + pInput->startRowIndex && i >= 0; i += step) {
+      if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
+        continue;
+      }
+
+      char*  d = (char*)pInputCol->pData + pInputCol->info.bytes * i;
+      GET_TYPED_DATA(v, double, pInputCol->info.type, d);
+
+      int32_t pos = pCtx->offset + numOfElems;
+      if (!pDerivInfo->valueSet) {  // initial value is not set yet
+        pDerivInfo->valueSet = true;
+      } else {
+        double r = ((v - pDerivInfo->prevValue) * pDerivInfo->tsWindow) / (tsList[i] - pDerivInfo->prevTs);
+        if (pDerivInfo->ignoreNegative && r < 0) {
+        } else {
+          colDataAppend(pOutput, pos, (const char*)&r, false);
+          if (pTsOutput != NULL) {
+            colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+          }
+          numOfElems++;
+        }
+      }
+
+      pDerivInfo->prevValue = v;
+      pDerivInfo->prevTs = tsList[i];
+    }
+  } else {
+
+  }
+
+  return numOfElems;
+}
\ No newline at end of file

source/libs/function/src/taggfunction.c

@@ -161,13 +161,13 @@ typedef struct SRateInfo {
   bool    isIRate;    // true for IRate functions, false for Rate functions
 } SRateInfo;
 
-typedef struct SDerivInfo {
-  double   prevValue;     // previous value
-  TSKEY    prevTs;        // previous timestamp
-  bool     ignoreNegative;// ignore the negative value
-  int64_t  tsWindow;      // time window for derivative
-  bool     valueSet;      // the value has been set already
-} SDerivInfo;
+//typedef struct SDerivInfo {
+//  double   prevValue;     // previous value
+//  TSKEY    prevTs;        // previous timestamp
+//  bool     ignoreNegative;// ignore the negative value
+//  int64_t  tsWindow;      // time window for derivative
+//  bool     valueSet;      // the value has been set already
+//} SDerivInfo;
 
 typedef struct SResPair {
   TSKEY  key;

source/libs/function/src/texpr.c

@@ -48,34 +48,6 @@ static void doExprTreeDestroy(tExprNode **pExpr, void (*fp)(void *)) {
   *pExpr = NULL;
 }
 
-bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp *param) {
-#if 0
-  //non-leaf nodes, recursively traverse the expression tree in the post-root order
-  if (pLeft->nodeType == TEXPR_BINARYEXPR_NODE && pRight->nodeType == TEXPR_BINARYEXPR_NODE) {
-    if (pExpr->_node.optr == LOGIC_COND_TYPE_OR) {  // or
-      if (exprTreeApplyFilter(pLeft, pItem, param)) {
-        return true;
-      }
-
-      // left child does not satisfy the query condition, try right child
-      return exprTreeApplyFilter(pRight, pItem, param);
-    } else {  // and
-      if (!exprTreeApplyFilter(pLeft, pItem, param)) {
-        return false;
-      }
-
-      return exprTreeApplyFilter(pRight, pItem, param);
-    }
-  }
-
-  // handle the leaf node
-  param->setupInfoFn(pExpr, param->pExtInfo);
-  return param->nodeFilterFn(pItem, pExpr->_node.info);
-#endif
-
-  return 0;
-}
-
 // TODO: these three functions should be made global
 static void* exception_calloc(size_t nmemb, size_t size) {
   void* p = taosMemoryCalloc(nmemb, size);
@@ -101,214 +73,3 @@ static UNUSED_FUNC char* exception_strdup(const char* str) {
   return p;
 }
 
-void buildFilterSetFromBinary(void **q, const char *buf, int32_t len) {
-  SBufferReader br = tbufInitReader(buf, len, false); 
-  uint32_t type  = tbufReadUint32(&br);     
-  SHashObj *pObj = taosHashInit(256, taosGetDefaultHashFunction(type), true, false);
-  
-//  taosHashSetEqualFp(pObj, taosGetDefaultEqualFunction(type));
-
-  int dummy = -1;
-  int32_t sz = tbufReadInt32(&br);
-  for (int32_t i = 0; i < sz; i++) {
-    if (type == TSDB_DATA_TYPE_BOOL || IS_SIGNED_NUMERIC_TYPE(type)) {
-      int64_t val = tbufReadInt64(&br); 
-      taosHashPut(pObj, (char *)&val, sizeof(val),  &dummy, sizeof(dummy));
-    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
-      uint64_t val = tbufReadUint64(&br); 
-      taosHashPut(pObj, (char *)&val, sizeof(val),  &dummy, sizeof(dummy));
-    }
-    else if (type == TSDB_DATA_TYPE_TIMESTAMP) {
-      int64_t val = tbufReadInt64(&br); 
-      taosHashPut(pObj, (char *)&val, sizeof(val),  &dummy, sizeof(dummy));
-    } else if (type == TSDB_DATA_TYPE_DOUBLE || type == TSDB_DATA_TYPE_FLOAT) {
-      double  val = tbufReadDouble(&br);
-      taosHashPut(pObj, (char *)&val, sizeof(val), &dummy, sizeof(dummy));
-    } else if (type == TSDB_DATA_TYPE_BINARY) {
-      size_t  t = 0;
-      const char *val = tbufReadBinary(&br, &t);
-      taosHashPut(pObj, (char *)val, t, &dummy, sizeof(dummy));
-    } else if (type == TSDB_DATA_TYPE_NCHAR) {
-      size_t  t = 0;
-      const char *val = tbufReadBinary(&br, &t);      
-      taosHashPut(pObj, (char *)val, t, &dummy, sizeof(dummy));
-    }
-  } 
-  *q = (void *)pObj;
-}
-
-void convertFilterSetFromBinary(void **q, const char *buf, int32_t len, uint32_t tType) {
-  SBufferReader br = tbufInitReader(buf, len, false); 
-  uint32_t sType  = tbufReadUint32(&br);     
-  SHashObj *pObj = taosHashInit(256, taosGetDefaultHashFunction(tType), true, false);
-  
-//  taosHashSetEqualFp(pObj, taosGetDefaultEqualFunction(tType));
-  
-  int dummy = -1;
-  SVariant tmpVar = {0};  
-  size_t  t = 0;
-  int32_t sz = tbufReadInt32(&br);
-  void *pvar = NULL;  
-  int64_t val = 0;
-  int32_t bufLen = 0;
-  if (IS_NUMERIC_TYPE(sType)) {
-    bufLen = 60;  // The maximum length of string that a number is converted to.
-  } else {
-    bufLen = 128;
-  }
-
-  char *tmp = taosMemoryCalloc(1, bufLen * TSDB_NCHAR_SIZE);
-    
-  for (int32_t i = 0; i < sz; i++) {
-    switch (sType) {
-    case TSDB_DATA_TYPE_BOOL:
-    case TSDB_DATA_TYPE_UTINYINT:
-    case TSDB_DATA_TYPE_TINYINT: {
-      *(uint8_t *)&val = (uint8_t)tbufReadInt64(&br); 
-      t = sizeof(val);
-      pvar = &val;
-      break;
-    }
-    case TSDB_DATA_TYPE_USMALLINT:
-    case TSDB_DATA_TYPE_SMALLINT: {
-      *(uint16_t *)&val = (uint16_t)tbufReadInt64(&br); 
-      t = sizeof(val);
-      pvar = &val;
-      break;
-    }
-    case TSDB_DATA_TYPE_UINT:
-    case TSDB_DATA_TYPE_INT: {
-      *(uint32_t *)&val = (uint32_t)tbufReadInt64(&br); 
-      t = sizeof(val);
-      pvar = &val;
-      break;
-    }
-    case TSDB_DATA_TYPE_TIMESTAMP:
-    case TSDB_DATA_TYPE_UBIGINT:
-    case TSDB_DATA_TYPE_BIGINT: {
-      *(uint64_t *)&val = (uint64_t)tbufReadInt64(&br); 
-      t = sizeof(val);
-      pvar = &val;
-      break;
-    }
-    case TSDB_DATA_TYPE_DOUBLE: {
-      *(double *)&val = tbufReadDouble(&br);
-      t = sizeof(val);
-      pvar = &val;
-      break;
-    }
-    case TSDB_DATA_TYPE_FLOAT: {
-      *(float *)&val = (float)tbufReadDouble(&br);
-      t = sizeof(val);
-      pvar = &val;
-      break;
-    }
-    case TSDB_DATA_TYPE_BINARY: {
-      pvar = (char *)tbufReadBinary(&br, &t);
-      break;
-    }
-    case TSDB_DATA_TYPE_NCHAR: {
-      pvar = (char *)tbufReadBinary(&br, &t);      
-      break;
-    }
-    default:
-      taosHashCleanup(pObj);
-      *q = NULL;
-      return;
-    }
-    
-    taosVariantCreateFromBinary(&tmpVar, (char *)pvar, t, sType);
-
-    if (bufLen < t) {
-      tmp = taosMemoryRealloc(tmp, t * TSDB_NCHAR_SIZE);
-      bufLen = (int32_t)t;
-    }
-
-    switch (tType) {
-      case TSDB_DATA_TYPE_BOOL:
-      case TSDB_DATA_TYPE_UTINYINT:
-      case TSDB_DATA_TYPE_TINYINT: {
-        if (taosVariantDump(&tmpVar, (char *)&val, tType, false)) {
-          goto err_ret;
-        }
-        pvar = &val;
-        t = sizeof(val);
-        break;
-      }
-      case TSDB_DATA_TYPE_USMALLINT:
-      case TSDB_DATA_TYPE_SMALLINT: {
-        if (taosVariantDump(&tmpVar, (char *)&val, tType, false)) {
-          goto err_ret;
-        }
-        pvar = &val;
-        t = sizeof(val);
-        break;
-      }
-      case TSDB_DATA_TYPE_UINT:
-      case TSDB_DATA_TYPE_INT: {
-        if (taosVariantDump(&tmpVar, (char *)&val, tType, false)) {
-          goto err_ret;
-        }
-        pvar = &val;
-        t = sizeof(val);
-        break;
-      }
-      case TSDB_DATA_TYPE_TIMESTAMP:
-      case TSDB_DATA_TYPE_UBIGINT:
-      case TSDB_DATA_TYPE_BIGINT: {
-        if (taosVariantDump(&tmpVar, (char *)&val, tType, false)) {
-          goto err_ret;
-        }
-        pvar = &val;
-        t = sizeof(val);
-        break;
-      }
-      case TSDB_DATA_TYPE_DOUBLE: {
-        if (taosVariantDump(&tmpVar, (char *)&val, tType, false)) {
-          goto err_ret;
-        }
-        pvar = &val;
-        t = sizeof(val);
-        break;
-      }
-      case TSDB_DATA_TYPE_FLOAT: {
-        if (taosVariantDump(&tmpVar, (char *)&val, tType, false)) {
-          goto err_ret;
-        }
-        pvar = &val;
-        t = sizeof(val);
-        break;
-      }
-      case TSDB_DATA_TYPE_BINARY: {
-        if (taosVariantDump(&tmpVar, tmp, tType, true)) {
-          goto err_ret;
-        }
-        t = varDataLen(tmp);
-        pvar = varDataVal(tmp);
-        break;
-      }
-      case TSDB_DATA_TYPE_NCHAR: {
-        if (taosVariantDump(&tmpVar, tmp, tType, true)) {
-          goto err_ret;
-        }
-        t = varDataLen(tmp);
-        pvar = varDataVal(tmp);        
-        break;
-      }
-      default:
-        goto err_ret;
-    }
-    
-    taosHashPut(pObj, (char *)pvar, t,  &dummy, sizeof(dummy));
-    taosVariantDestroy(&tmpVar);
-    memset(&tmpVar, 0, sizeof(tmpVar));
-  } 
-
-  *q = (void *)pObj;
-  pObj = NULL;
-  
-err_ret:  
-  taosVariantDestroy(&tmpVar);
-  taosHashCleanup(pObj);
-  taosMemoryFreeClear(tmp);
-}