feat(query): add to_iso8601 function

TD-14243

include/libs/scalar/scalar.h

@@ -73,6 +73,9 @@ int32_t substrFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
 /* Conversion functions */
 int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
 
+/* Time related functions */
+int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+
 bool getTimePseudoFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 
 int32_t winStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);

include/util/tdef.h

@@ -94,6 +94,11 @@ extern const int32_t TYPE_BYTES[15];
 #define TSDB_TIME_PRECISION_MICRO_STR "us"
 #define TSDB_TIME_PRECISION_NANO_STR  "ns"
 
+#define TSDB_TIME_PRECISION_SEC_DIGITS   10
+#define TSDB_TIME_PRECISION_MILLI_DIGITS 13
+#define TSDB_TIME_PRECISION_MICRO_DIGITS 16
+#define TSDB_TIME_PRECISION_NANO_DIGITS  19
+
 #define TSDB_INFORMATION_SCHEMA_DB            "information_schema"
 #define TSDB_INS_TABLE_DNODES                 "dnodes"
 #define TSDB_INS_TABLE_MNODES                 "mnodes"

source/libs/function/src/builtins.c

@@ -393,6 +393,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .sprocessFunc = castFunction,
     .finalizeFunc = NULL
   },
+  {
+    .name = "to_iso8601",
+    .type = FUNCTION_TYPE_TO_ISO8601,
+    .classification = FUNC_MGT_SCALAR_FUNC,
+    .checkFunc    = checkAndGetResultType,
+    .getEnvFunc   = NULL,
+    .initFunc     = NULL,
+    .sprocessFunc = toISO8601Function,
+    .finalizeFunc = NULL
+  },
   {
     .name = "_rowts",
     .type = FUNCTION_TYPE_ROWTS,
@@ -609,6 +619,9 @@ int32_t checkAndGetResultType(SFunctionNode* pFunc) {
       pFunc->node.resType = (SDataType) { .bytes = paraBytes, .type = paraType};
       break;
     }
+    case FUNCTION_TYPE_TO_ISO8601: {
+      pFunc->node.resType = (SDataType) { .bytes = 64, .type = TSDB_DATA_TYPE_BINARY};
+    }
 
     case FUNCTION_TYPE_TBNAME: {
       // todo

source/libs/scalar/src/sclfunc.c

@@ -801,6 +801,77 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
+  int32_t type = GET_PARAM_TYPE(pInput);
+  if (type != TSDB_DATA_TYPE_BIGINT && type != TSDB_DATA_TYPE_TIMESTAMP) {
+    return TSDB_CODE_FAILED;
+  }
+
+  if (inputNum != 1) {
+    return TSDB_CODE_FAILED;
+  }
+
+  char *input  = pInput[0].columnData->pData;
+  for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
+    if (colDataIsNull_s(pInput[0].columnData, i)) {
+      colDataAppendNULL(pOutput->columnData, i);
+      continue;
+    }
+
+    char fraction[20] = {0};
+    bool hasFraction = false;
+    NUM_TO_STRING(type, input, sizeof(fraction), fraction);
+    int32_t tsDigits = (int32_t)strlen(fraction);
+
+    char buf[64] = {0};
+    int64_t timeVal;
+    GET_TYPED_DATA(timeVal, int64_t, type, input);
+    if (tsDigits > TSDB_TIME_PRECISION_SEC_DIGITS) {
+      if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
+        timeVal = timeVal / 1000;
+      } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
+        timeVal = timeVal / (1000 * 1000);
+      } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
+        timeVal = timeVal / (1000 * 1000 * 1000);
+      } else {
+        assert(0);
+      }
+      hasFraction = true;
+      memmove(fraction, fraction + TSDB_TIME_PRECISION_SEC_DIGITS, TSDB_TIME_PRECISION_SEC_DIGITS);
+    }
+
+    struct tm *tmInfo = localtime((const time_t *)&timeVal);
+    strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%S%z", tmInfo);
+    int32_t len = (int32_t)strlen(buf);
+
+    if (hasFraction) {
+      int32_t fracLen = (int32_t)strlen(fraction) + 1;
+      char *tzInfo = strchr(buf, '+');
+      if (tzInfo) {
+        memmove(tzInfo + fracLen, tzInfo, strlen(tzInfo));
+      } else {
+        tzInfo = strchr(buf, '-');
+        memmove(tzInfo + fracLen, tzInfo, strlen(tzInfo));
+      }
+
+      char tmp[32];
+      sprintf(tmp, ".%s", fraction);
+      memcpy(tzInfo, tmp, fracLen);
+      len += fracLen;
+    }
+
+    memmove(buf + VARSTR_HEADER_SIZE, buf, len);
+    varDataSetLen(buf, len);
+
+    colDataAppend(pOutput->columnData, i, buf, false);
+    input   += tDataTypes[type].bytes;
+  }
+
+  pOutput->numOfRows = pInput->numOfRows;
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t atanFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
   return doScalarFunctionUnique(pInput, inputNum, pOutput, atan);
 }