Merge pull request #12605 from taosdata/feature/udf

fix: trigger ci test

include/util/taoserror.h

@@ -668,6 +668,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_UDF_INVALID_STATE             TAOS_DEF_ERROR_CODE(0, 0x2906)
 #define TSDB_CODE_UDF_INVALID_INPUT             TAOS_DEF_ERROR_CODE(0, 0x2907)
 #define TSDB_CODE_UDF_NO_FUNC_HANDLE            TAOS_DEF_ERROR_CODE(0, 0x2908)
+#define TSDB_CODE_UDF_INVALID_BUFSIZE           TAOS_DEF_ERROR_CODE(0, 0x2909)
+#define TSDB_CODE_UDF_INVALID_OUTPUT_TYPE       TAOS_DEF_ERROR_CODE(0, 0x290A)
 
 #define TSDB_CODE_SML_INVALID_PROTOCOL_TYPE     TAOS_DEF_ERROR_CODE(0, 0x3000)
 #define TSDB_CODE_SML_INVALID_PRECISION_TYPE    TAOS_DEF_ERROR_CODE(0, 0x3001)

source/libs/function/src/tudf.c

@@ -1605,7 +1605,12 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
     return false;
   }
   udfRes->interResNum = buf.numOfResult;
-  memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
+  if (buf.bufLen <= session->bufSize) {
+    memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
+  } else {
+    fnError("udfc inter buf size %d is greater than function bufSize %d", buf.bufLen, session->bufSize);
+    return false;
+  }
   freeUdfInterBuf(&buf);
   return true;
 }
@@ -1655,7 +1660,12 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
     newState.numOfResult = 0;
   } else {
     udfRes->interResNum = newState.numOfResult;
-    memcpy(udfRes->interResBuf, newState.buf, newState.bufLen);
+    if (newState.bufLen <= session->bufSize) {
+      memcpy(udfRes->interResBuf, newState.buf, newState.bufLen);
+    } else {
+      fnError("udfc inter buf size %d is greater than function bufSize %d", newState.bufLen, session->bufSize);
+      udfCode = TSDB_CODE_UDF_INVALID_BUFSIZE;
+    }
   }
   if (newState.numOfResult == 1 || state.numOfResult == 1) {
     GET_RES_INFO(pCtx)->numOfRes = 1;
@@ -1688,9 +1698,15 @@ int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock) {
     fnError("udfAggFinalize error. doCallUdfAggFinalize step. udf code:%d", udfCallCode);
     GET_RES_INFO(pCtx)->numOfRes = 0;
   } else {
-    memcpy(udfRes->finalResBuf, resultBuf.buf, session->outputLen);
-    udfRes->finalResNum = resultBuf.numOfResult;
-    GET_RES_INFO(pCtx)->numOfRes = udfRes->finalResNum;
+    if (resultBuf.bufLen <= session->outputLen) {
+      memcpy(udfRes->finalResBuf, resultBuf.buf, session->outputLen);
+      udfRes->finalResNum = resultBuf.numOfResult;
+      GET_RES_INFO(pCtx)->numOfRes = udfRes->finalResNum;
+    } else {
+      fnError("udfc inter buf size %d is greater than function output size %d", resultBuf.bufLen, session->outputLen);
+      GET_RES_INFO(pCtx)->numOfRes = 0;
+      udfCallCode = TSDB_CODE_UDF_INVALID_OUTPUT_TYPE;
+    }
   }
 
   freeUdfInterBuf(&resultBuf);

source/libs/function/src/udfd.c

@@ -616,9 +616,8 @@ void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
 }
 
 void udfdOnNewConnection(uv_stream_t *server, int status) {
-  fnDebug("new connection");
   if (status < 0) {
-    // TODO
+    fnError("udfd new connection error. code: %s", uv_strerror(status));
     return;
   }
 

source/util/src/terror.c

@@ -470,6 +470,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_UDF_LOAD_UDF_FAILURE,         "udf load failure")
 TAOS_DEFINE_ERROR(TSDB_CODE_UDF_INVALID_STATE,            "udf invalid state")
 TAOS_DEFINE_ERROR(TSDB_CODE_UDF_INVALID_INPUT,            "udf invalid function input")
 TAOS_DEFINE_ERROR(TSDB_CODE_UDF_NO_FUNC_HANDLE,           "udf no function handle")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_INVALID_BUFSIZE,          "udf invalid bufsize")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_INVALID_OUTPUT_TYPE,      "udf invalid output type")
+
 //schemaless
 TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_PROTOCOL_TYPE,    "Invalid line protocol type")
 TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_PRECISION_TYPE,   "Invalid timestamp precision type")