Merge pull request #12552 from taosdata/feature/3.0_glzhao

feat(query): add tail function

source/libs/function/inc/builtinsimpl.h

@@ -105,7 +105,12 @@ int32_t mavgFunction(SqlFunctionCtx* pCtx);
 bool getSampleFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool sampleFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t sampleFunction(SqlFunctionCtx* pCtx);
-int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+//int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+
+bool getTailFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+bool tailFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
+int32_t tailFunction(SqlFunctionCtx* pCtx);
+int32_t tailFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 
 bool getSelectivityFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 

source/libs/function/src/builtins.c

@@ -386,6 +386,35 @@ static int32_t translateSample(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateTail(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList);
+  if (2 != numOfParams && 3 != numOfParams) {
+    return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
+  if (QUERY_NODE_COLUMN != nodeType(pPara)) {
+    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
+                           "The input parameter of TAIL function can only be column");
+  }
+
+  for (int32_t i = 1; i < numOfParams; ++i) {
+    uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, i))->resType.type;
+    if (!IS_INTEGER_TYPE(paraType)) {
+      return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+    }
+  }
+
+  SExprNode* pCol = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0);
+  uint8_t colType = pCol->resType.type;
+  if (IS_VAR_DATA_TYPE(colType)) {
+    pFunc->node.resType = (SDataType){.bytes = pCol->resType.bytes, .type = colType};
+  } else {
+    pFunc->node.resType = (SDataType){.bytes = tDataTypes[colType].bytes, .type = colType};
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t translateLastRow(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   // todo
   return TSDB_CODE_SUCCESS;
@@ -850,6 +879,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .processFunc  = sampleFunction,
     .finalizeFunc = NULL
   },
+  {
+    .name = "tail",
+    .type = FUNCTION_TYPE_TAIL,
+    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .translateFunc = translateTail,
+    .getEnvFunc   = getTailFuncEnv,
+    .initFunc     = tailFunctionSetup,
+    .processFunc  = tailFunction,
+    .finalizeFunc = tailFinalize
+  },
   {
     .name = "abs",
     .type = FUNCTION_TYPE_ABS,

source/libs/function/src/builtinsimpl.c

@@ -18,12 +18,15 @@
 #include "function.h"
 #include "querynodes.h"
 #include "taggfunction.h"
+#include "tcompare.h"
 #include "tdatablock.h"
 #include "tpercentile.h"
 
 #define HISTOGRAM_MAX_BINS_NUM   1000
 #define MAVG_MAX_POINTS_NUM      1000
 #define SAMPLE_MAX_POINTS_NUM    1000
+#define TAIL_MAX_POINTS_NUM      100
+#define TAIL_MAX_OFFSET          100
 
 typedef struct SSumRes {
   union {
@@ -161,6 +164,21 @@ typedef struct SSampleInfo {
   int64_t *timestamp;
 } SSampleInfo;
 
+typedef struct STailItem {
+  int64_t timestamp;
+  bool    isNull;
+  char    data[];
+} STailItem;
+
+typedef struct STailInfo {
+  int32_t   numOfPoints;
+  int32_t   numAdded;
+  int32_t   offset;
+  uint8_t   colType;
+  int16_t   colBytes;
+  STailItem **pItems;
+} STailInfo;
+
 #define SET_VAL(_info, numOfElem, res) \
   do {                                 \
     if ((numOfElem) <= 0) {            \
@@ -3107,7 +3125,7 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
 
   int32_t startOffset = pCtx->offset;
   for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
-    if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
+    if (colDataIsNull_s(pInputCol, i)) {
       //colDataAppendNULL(pOutput, i);
       continue;
     }
@@ -3141,3 +3159,132 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
 //
 //  return pResInfo->numOfRes;
 //}
+
+
+bool getTailFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
+  SColumnNode* pCol = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0);
+  SValueNode*  pVal = (SValueNode*)nodesListGetNode(pFunc->pParameterList, 1);
+  int32_t numOfPoints = pVal->datum.i;
+  pEnv->calcMemSize = sizeof(STailInfo) + numOfPoints * (POINTER_BYTES + sizeof(STailItem) + pCol->node.resType.bytes);
+  return true;
+}
+
+bool tailFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo *pResultInfo) {
+  if (!functionSetup(pCtx, pResultInfo)) {
+    return false;
+  }
+
+  STailInfo *pInfo = GET_ROWCELL_INTERBUF(pResultInfo);
+  pInfo->numAdded = 0;
+  pInfo->numOfPoints = pCtx->param[1].param.i;
+  if (pCtx->numOfParams == 4) {
+    pInfo->offset = pCtx->param[2].param.i;
+  } else {
+    pInfo->offset = 0;
+  }
+  pInfo->colType = pCtx->resDataInfo.type;
+  pInfo->colBytes = pCtx->resDataInfo.bytes;
+  if ((pInfo->numOfPoints < 1 || pInfo->numOfPoints > TAIL_MAX_POINTS_NUM) ||
+      (pInfo->numOfPoints < 0 || pInfo->numOfPoints > TAIL_MAX_OFFSET)) {
+    return false;
+  }
+
+  pInfo->pItems = (STailItem **)((char *)pInfo + sizeof(STailInfo));
+  char *pItem = (char *)pInfo->pItems + pInfo->numOfPoints * POINTER_BYTES;
+
+  size_t unitSize = sizeof(STailItem) + pInfo->colBytes;
+  for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
+    pInfo->pItems[i] = (STailItem *)(pItem + i * unitSize);
+    pInfo->pItems[i]->isNull = false;
+  }
+
+  return true;
+}
+
+static void tailAssignResult(STailItem* pItem, char *data, int32_t colBytes, TSKEY ts, bool isNull) {
+  pItem->timestamp = ts;
+  if (isNull) {
+    pItem->isNull = true;
+  } else {
+    memcpy(pItem->data, data, colBytes);
+  }
+}
+
+static int32_t tailCompFn(const void *p1, const void *p2, const void *param) {
+  STailItem *d1 = *(STailItem **)p1;
+  STailItem *d2 = *(STailItem **)p2;
+  return compareInt64Val(&d1->timestamp, &d2->timestamp);
+}
+
+static void doTailAdd(STailInfo* pInfo, char *data, TSKEY ts, bool isNull) {
+  STailItem **pList = pInfo->pItems;
+  if (pInfo->numAdded < pInfo->numOfPoints) {
+    tailAssignResult(pList[pInfo->numAdded], data, pInfo->colBytes, ts, isNull);
+    taosheapsort((void *)pList, sizeof(STailItem **), pInfo->numAdded + 1, NULL, tailCompFn, 0);
+    pInfo->numAdded++;
+  } else if (pList[0]->timestamp < ts) {
+    tailAssignResult(pList[0], data, pInfo->colBytes, ts, isNull);
+    taosheapadjust((void *)pList, sizeof(STailItem **), 0, pInfo->numOfPoints - 1, NULL, tailCompFn, NULL, 0);
+  }
+}
+
+int32_t tailFunction(SqlFunctionCtx* pCtx) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  STailInfo*         pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+
+  SInputColumnInfoData* pInput = &pCtx->input;
+  TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
+
+  SColumnInfoData* pInputCol = pInput->pData[0];
+  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
+  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
+
+  int32_t startOffset = pCtx->offset;
+  if (pInfo->offset >= pInput->numOfRows) {
+    return 0;
+  } else {
+    pInfo->numOfPoints = MIN(pInfo->numOfPoints, pInput->numOfRows - pInfo->offset);
+  }
+  for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex - pInfo->offset; i += 1) {
+
+    char* data = colDataGetData(pInputCol, i);
+    doTailAdd(pInfo, data, tsList[i], colDataIsNull_s(pInputCol, i));
+  }
+
+  taosqsort(pInfo->pItems, pInfo->numOfPoints, POINTER_BYTES, NULL, tailCompFn);
+
+  for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
+    int32_t pos = startOffset + i;
+    STailItem *pItem = pInfo->pItems[i];
+    if (pItem->isNull) {
+      colDataAppendNULL(pOutput, pos);
+    } else {
+      colDataAppend(pOutput, pos, pItem->data, false);
+    }
+  }
+
+  return pInfo->numOfPoints;
+}
+
+int32_t tailFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
+  STailInfo*                pInfo = GET_ROWCELL_INTERBUF(pEntryInfo);
+  pEntryInfo->complete = true;
+
+  int32_t type = pCtx->input.pData[0]->info.type;
+  int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
+
+  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
+
+  // todo assign the tag value and the corresponding row data
+  int32_t currentRow = pBlock->info.rows;
+  for (int32_t i = 0; i < pEntryInfo->numOfRes; ++i) {
+    STailItem *pItem = pInfo->pItems[i];
+    colDataAppend(pCol, currentRow, pItem->data, false);
+
+    //setSelectivityValue(pCtx, pBlock, &pInfo->pItems[i].tuplePos, currentRow);
+    currentRow += 1;
+  }
+
+  return pEntryInfo->numOfRes;
+}