User-Defined Functions (UDF)

The built-in functions of TDengine may not be sufficient for the use cases of every application. In this case, you can define custom functions for use in TDengine queries. These are known as user-defined functions (UDF). A user-defined function takes one column of data or the result of a subquery as its input.

TDengine supports user-defined functions written in C or C++. This document describes the usage of user-defined functions.

User-defined functions can be scalar functions or aggregate functions. Scalar functions, such as abs, sin, and concat, output a value for every row of data. Aggregate functions, such as avg and max output one value for multiple rows of data.

When you create a user-defined function, you must implement standard interface functions:

  • For scalar functions, implement the scalarfn interface function.
  • For aggregate functions, implement the aggfn_start, aggfn, and aggfn_finish interface functions.
  • To initialize your function, implement the udf_init function. To terminate your function, implement the udf_destroy function.

There are strict naming conventions for these interface functions. The names of the start, finish, init, and destroy interfaces must be <udf-name>_start, <udf-name>_finish, <udf-name>_init, and <udf-name>_destroy, respectively. Replace scalarfn, aggfn, and udf with the name of your user-defined function.

Implementing a Scalar Function

The implementation of a scalar function is described as follows:

  1. #include "taos.h"
  2. #include "taoserror.h"
  3. #include "taosudf.h"
  5. // initialization function. if no initialization, we can skip definition of it. The initialization function shall be concatenation of the udf name and _init suffix
  6. // @return error number defined in taoserror.h
  7. int32_t scalarfn_init() {
  8.     // initialization.
  9.     return TSDB_CODE_SUCCESS;
  10. }
  12. // scalar function main computation function
  13. // @param inputDataBlock, input data block composed of multiple columns with each column defined by SUdfColumn
  14. // @param resultColumn, output column
  15. // @return error number defined in taoserror.h
  16. int32_t scalarfn(SUdfDataBlock* inputDataBlock, SUdfColumn* resultColumn) {
  17.     // read data from inputDataBlock and process, then output to resultColumn.
  18.     return TSDB_CODE_SUCCESS;
  19. }
  21. // cleanup function. if no cleanup related processing, we can skip definition of it. The destroy function shall be concatenation of the udf name and _destroy suffix.
  22. // @return error number defined in taoserror.h
  23. int32_t scalarfn_destroy() {
  24.     // clean up
  25.     return TSDB_CODE_SUCCESS;
  26. }

Replace scalarfn with the name of your function.

Implementing an Aggregate Function

The implementation of an aggregate function is described as follows:

  1. #include "taos.h"
  2. #include "taoserror.h"
  3. #include "taosudf.h"
  5. // Initialization function. if no initialization, we can skip definition of it. The initialization function shall be concatenation of the udf name and _init suffix
  6. // @return error number defined in taoserror.h
  7. int32_t aggfn_init() {
  8.     // initialization.
  9.     return TSDB_CODE_SUCCESS;
  10. }
  12. // aggregate start function. The intermediate value or the state(@interBuf) is initialized in this function. The function name shall be concatenation of udf name and _start suffix
  13. // @param interbuf intermediate value to intialize
  14. // @return error number defined in taoserror.h
  15. int32_t aggfn_start(SUdfInterBuf* interBuf) {
  16.     // initialize intermediate value in interBuf
  17.     return TSDB_CODE_SUCESS;
  18. }
  20. // aggregate reduce function. This function aggregate old state(@interbuf) and one data bock(inputBlock) and output a new state(@newInterBuf).
  21. // @param inputBlock input data block
  22. // @param interBuf old state
  23. // @param newInterBuf new state
  24. // @return error number defined in taoserror.h
  25. int32_t aggfn(SUdfDataBlock* inputBlock, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
  26.     // read from inputBlock and interBuf and output to newInterBuf
  27.     return TSDB_CODE_SUCCESS;
  28. }
  30. // aggregate function finish function. This function transforms the intermediate value(@interBuf) into the final output(@result). The function name must be concatenation of aggfn and _finish suffix.
  31. // @interBuf : intermediate value
  32. // @result: final result
  33. // @return error number defined in taoserror.h
  34. int32_t int32_t aggfn_finish(SUdfInterBuf* interBuf, SUdfInterBuf *result) {
  35.     // read data from inputDataBlock and process, then output to result
  36.     return TSDB_CODE_SUCCESS;
  37. }
  39. // cleanup function. if no cleanup related processing, we can skip definition of it. The destroy function shall be concatenation of the udf name and _destroy suffix.
  40. // @return error number defined in taoserror.h
  41. int32_t aggfn_destroy() {
  42.     // clean up
  43.     return TSDB_CODE_SUCCESS;
  44. }

Replace aggfn with the name of your function.

Interface Functions

There are strict naming conventions for interface functions. The names of the start, finish, init, and destroy interfaces must be <udf-name>_start, <udf-name>_finish, <udf-name>_init, and <udf-name>_destroy, respectively. Replace scalarfn, aggfn, and udf with the name of your user-defined function.

Interface functions return a value that indicates whether the operation was successful. If an operation fails, the interface function returns an error code. Otherwise, it returns TSDB_CODE_SUCCESS. The error codes are defined in taoserror.h and in the common API error codes in taos.h. For example, TSDB_CODE_UDF_INVALID_INPUT indicates invalid input. TSDB_CODE_OUT_OF_MEMORY indicates insufficient memory.

For information about the parameters for interface functions, see Data Model

Interfaces for Scalar Functions

int32_t scalarfn(SUdfDataBlock* inputDataBlock, SUdfColumn *resultColumn)

Replace scalarfn with the name of your function. This function performs scalar calculations on data blocks. You can configure a value through the parameters in the resultColumn structure.

The parameters in the function are defined as follows:

  • inputDataBlock: The data block to input.
  • resultColumn: The column to output. The column to output.

Interfaces for Aggregate Functions

int32_t aggfn_start(SUdfInterBuf *interBuf)

int32_t aggfn(SUdfDataBlock* inputBlock, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf)

int32_t aggfn_finish(SUdfInterBuf* interBuf, SUdfInterBuf *result)

Replace aggfn with the name of your function. In the function, aggfn_start is called to generate a result buffer. Data is then divided between multiple blocks, and aggfn is called on each block to update the result. Finally, aggfn_finish is called to generate final results from the intermediate results. The final result contains only one or zero data points.

The parameters in the function are defined as follows:

  • interBuf: The intermediate result buffer.
  • inputBlock: The data block to input.
  • newInterBuf: The new intermediate result buffer.
  • result: The final result.

Initializing and Terminating User-Defined Functions

int32_t udf_init()

int32_t udf_destroy()

Replace udfwith the name of your function. udf_init initializes the function. udf_destroy terminates the function. If it is not necessary to initialize your function, udf_init is not required. If it is not necessary to terminate your function, udf_destroy is not required.

Data Structure of User-Defined Functions

  1. typedef struct SUdfColumnMeta {
  2.   int16_t type;
  3.   int32_t bytes;
  4.   uint8_t precision;
  5.   uint8_t scale;
  6. } SUdfColumnMeta;
  8. typedef struct SUdfColumnData {
  9.   int32_t numOfRows;
  10.   int32_t rowsAlloc;
  11.   union {
  12.     struct {
  13.       int32_t nullBitmapLen;
  14.       char   *nullBitmap;
  15.       int32_t dataLen;
  16.       char   *data;
  17.     } fixLenCol;
  19.     struct {
  20.       int32_t varOffsetsLen;
  21.       int32_t   *varOffsets;
  22.       int32_t payloadLen;
  23.       char   *payload;
  24.       int32_t payloadAllocLen;
  25.     } varLenCol;
  26.   };
  27. } SUdfColumnData;
  29. typedef struct SUdfColumn {
  30.   SUdfColumnMeta colMeta;
  31.   bool           hasNull;
  32.   SUdfColumnData colData;
  33. } SUdfColumn;
  35. typedef struct SUdfDataBlock {
  36.   int32_t numOfRows;
  37.   int32_t numOfCols;
  38.   SUdfColumn **udfCols;
  39. } SUdfDataBlock;
  41. typedef struct SUdfInterBuf {
  42.   int32_t bufLen;
  43.   char* buf;
  44.   int8_t numOfResult; //zero or one
  45. } SUdfInterBuf;

The data structure is described as follows:

  • The SUdfDataBlock block includes the number of rows (numOfRows) and number of columns (numCols). udfCols[i] (0 <= i <= numCols-1) indicates that each column is of type SUdfColumn.
  • SUdfColumn includes the definition of the data type of the column (colMeta) and the data in the column (colData).
  • The member definitions of SUdfColumnMeta are the same as the data type definitions in taos.h.
  • The data in SUdfColumnData can become longer. varLenCol indicates variable-length data, and fixLenCol indicates fixed-length data.
  • SUdfInterBuf defines the intermediate structure buffer and the number of results in the buffer numOfResult.

Additional functions are defined in taosudf.h to make it easier to work with these structures.

Compile UDF

To use your user-defined function in TDengine, first compile it to a dynamically linked library (DLL).

For example, the sample UDF bit_and.c can be compiled into a DLL as follows:

  1. gcc -g -O0 -fPIC -shared bit_and.c -o libbitand.so

The generated DLL file libbitand.so can now be used to implement your function. Note: GCC 7.5 or later is required.

Manage and Use User-Defined Functions

After compiling your function into a DLL, you add it to TDengine. For more information, see User-Defined Functions.

Sample Code

Sample scalar function: bit_and

The bit_and function implements bitwise addition for multiple columns. If there is only one column, the column is returned. The bit_and function ignores null values.

bit_and.c

  1. #include <string.h>
  2. #include <stdlib.h>
  3. #include <stdio.h>
  4. #include "taosudf.h"
  7. DLL_EXPORT int32_t bit_and_init() {
  8.     return 0;
  9. }
  11. DLL_EXPORT int32_t bit_and_destroy() {
  12.     return 0;
  13. }
  15. DLL_EXPORT int32_t bit_and(SUdfDataBlock* block, SUdfColumn *resultCol) {
  17.     if (block->numOfCols < 2) {
  18.         return TSDB_CODE_UDF_INVALID_INPUT;
  19.     }
  21.     for (int32_t i = 0; i < block->numOfCols; ++i) {
  22.         SUdfColumn* col = block->udfCols[i];
  23.         if (!(col->colMeta.type == TSDB_DATA_TYPE_INT)) {
  24.             return TSDB_CODE_UDF_INVALID_INPUT;
  25.         }
  26.     }
  28.     SUdfColumnMeta *meta = &resultCol->colMeta;
  29.     meta->bytes = 4;
  30.     meta->type = TSDB_DATA_TYPE_INT;
  31.     meta->scale = 0;
  32.     meta->precision = 0;
  35.     SUdfColumnData *resultData = &resultCol->colData;
  37.     resultData->numOfRows = block->numOfRows;
  39.     for (int32_t i = 0; i < resultData->numOfRows; ++i) {
  40.         if (udfColDataIsNull(block->udfCols[0], i)) {
  41.             udfColDataSetNull(resultCol, i);
  42.             continue;
  43.         }
  44.         int32_t result = *(int32_t*)udfColDataGetData(block->udfCols[0], i);
  45.         int j = 1;
  46.         for (; j < block->numOfCols; ++j) {
  47.             if (udfColDataIsNull(block->udfCols[j], i)) {
  48.                 udfColDataSetNull(resultCol, i);
  49.                 break;
  50.             }
  52.             char* colData = udfColDataGetData(block->udfCols[j], i);
  53.             result &= *(int32_t*)colData;
  54.         }
  55.         if (j == block->numOfCols) {
  56.             udfColDataSet(resultCol, i, (char*)&result, false);
  57.         }
  59.     }
  60.     return TSDB_CODE_SUCCESS;
  61. }

view source code

Sample aggregate function: l2norm

The l2norm function finds the second-order norm for all data in the input column. This squares the values, takes a cumulative sum, and finds the square root.

l2norm.c

  1. #include <string.h>
  2. #include <stdlib.h>
  3. #include <stdio.h>
  4. #include <math.h>
  6. #include "taosudf.h"
  8. DLL_EXPORT int32_t l2norm_init() {
  9.   return 0;
  10. }
  12. DLL_EXPORT int32_t l2norm_destroy() {
  13.   return 0;
  14. }
  16. DLL_EXPORT int32_t l2norm_start(SUdfInterBuf *buf) {
  17.   *(int64_t*)(buf->buf) = 0;
  18.   buf->bufLen = sizeof(double);
  19.   buf->numOfResult = 0;
  20.   return 0;
  21. }
  23. DLL_EXPORT int32_t l2norm(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
  24.   double sumSquares = *(double*)interBuf->buf;
  25.   int8_t numNotNull = 0;
  26.   for (int32_t i = 0; i < block->numOfCols; ++i) {
  27.     SUdfColumn* col = block->udfCols[i];
  28.     if (!(col->colMeta.type == TSDB_DATA_TYPE_INT || 
  29.           col->colMeta.type == TSDB_DATA_TYPE_DOUBLE)) {
  30.       return TSDB_CODE_UDF_INVALID_INPUT;
  31.     }
  32.   }
  33.   for (int32_t i = 0; i < block->numOfCols; ++i) {
  34.     for (int32_t j = 0; j < block->numOfRows; ++j) {
  35.       SUdfColumn* col = block->udfCols[i];
  36.       if (udfColDataIsNull(col, j)) {
  37.         continue;
  38.       }
  39.       switch (col->colMeta.type) {
  40.         case TSDB_DATA_TYPE_INT: {
  41.           char* cell = udfColDataGetData(col, j);
  42.           int32_t num = *(int32_t*)cell;
  43.           sumSquares += (double)num * num;
  44.           break;
  45.         }
  46.         case TSDB_DATA_TYPE_DOUBLE: {
  47.           char* cell = udfColDataGetData(col, j);
  48.           double num = *(double*)cell;
  49.           sumSquares += num * num;
  50.           break;
  51.         }
  52.         default: 
  53.           break;
  54.       }
  55.       ++numNotNull;
  56.     }
  57.   }
  59.   *(double*)(newInterBuf->buf) = sumSquares;
  60.   newInterBuf->bufLen = sizeof(double);
  62.   if (interBuf->numOfResult == 0 && numNotNull == 0) {
  63.     newInterBuf->numOfResult = 0;
  64.   } else {
  65.     newInterBuf->numOfResult = 1;
  66.   }
  67.   return 0;
  68. }
  70. DLL_EXPORT int32_t l2norm_finish(SUdfInterBuf* buf, SUdfInterBuf *resultData) {
  71.   if (buf->numOfResult == 0) {
  72.     resultData->numOfResult = 0;
  73.     return 0;
  74.   }
  75.   double sumSquares = *(double*)(buf->buf);
  76.   *(double*)(resultData->buf) = sqrt(sumSquares);
  77.   resultData->bufLen = sizeof(double);
  78.   resultData->numOfResult = 1;
  79.   return 0;
  80. }

view source code