Merge pull request #14399 from taosdata/enh/last_row_function

feat(query): add last_row function without cache

include/libs/function/functionMgt.h

@@ -34,6 +34,7 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_ELAPSED,
   FUNCTION_TYPE_IRATE,
   FUNCTION_TYPE_LAST_ROW,
+  FUNCTION_TYPE_LAST_ROWT, //TODO: removed
   FUNCTION_TYPE_MAX,
   FUNCTION_TYPE_MIN,
   FUNCTION_TYPE_MODE,

source/libs/function/inc/builtinsimpl.h

@@ -117,6 +117,9 @@ int32_t firstCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
 int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
 int32_t getFirstLastInfoSize(int32_t resBytes);
 
+int32_t lastRowFunction(SqlFunctionCtx *pCtx);
+int32_t lastRowFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+
 bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
 bool getTopBotMergeFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
 bool topBotFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);

source/libs/function/src/builtins.c

@@ -1953,16 +1953,13 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   },
   {
     .name = "last_row",
-    .type = FUNCTION_TYPE_LAST_ROW,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .type = FUNCTION_TYPE_LAST_ROWT,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateFirstLast,
     .getEnvFunc   = getFirstLastFuncEnv,
     .initFunc     = functionSetup,
-    .processFunc  = lastFunction,
-    .finalizeFunc = firstLastFinalize,
-    .pPartialFunc = "_last_partial",
-    .pMergeFunc   = "_last_merge",
-    .combineFunc  = lastCombine,
+    .processFunc  = lastRowFunction,
+    .finalizeFunc = lastRowFinalize,
   },
   {
     .name = "_cache_last_row",

source/libs/function/src/builtinsimpl.c

@@ -80,6 +80,7 @@ typedef struct STopBotRes {
 
 typedef struct SFirstLastRes {
   bool    hasResult;
+  bool    isNull;  //used for last_row function only
   int32_t bytes;
   char    buf[];
 } SFirstLastRes;
@@ -2763,6 +2764,113 @@ int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
+  int32_t numOfElems = 0;
+
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SFirstLastRes*       pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+
+  SInputColumnInfoData* pInput = &pCtx->input;
+  SColumnInfoData*      pInputCol = pInput->pData[0];
+
+  int32_t type = pInputCol->info.type;
+  int32_t bytes = pInputCol->info.bytes;
+  pInfo->bytes = bytes;
+
+  SColumnDataAgg* pColAgg = (pInput->colDataAggIsSet) ? pInput->pColumnDataAgg[0] : NULL;
+
+  TSKEY startKey = getRowPTs(pInput->pPTS, 0);
+  TSKEY endKey = getRowPTs(pInput->pPTS, pInput->totalRows - 1);
+
+  int32_t blockDataOrder = (startKey <= endKey) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
+
+  if (blockDataOrder == TSDB_ORDER_ASC) {
+    for (int32_t i = pInput->numOfRows + pInput->startRowIndex - 1; i >= pInput->startRowIndex; --i) {
+      char* data = colDataGetData(pInputCol, i);
+      TSKEY cts = getRowPTs(pInput->pPTS, i);
+      if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf) < cts) {
+        if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
+          pInfo->isNull = true;
+        } else {
+          pInfo->isNull = false;
+          if (IS_VAR_DATA_TYPE(type)) {
+            bytes = varDataTLen(data);
+            pInfo->bytes = bytes;
+          }
+          memcpy(pInfo->buf + sizeof(TSKEY), data, bytes);
+        }
+        *(TSKEY*)(pInfo->buf) = cts;
+        numOfElems++;
+        //handle selectivity
+        if (pCtx->subsidiaries.num > 0) {
+          STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
+          if (!pInfo->hasResult) {
+            saveTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          } else {
+            copyTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          }
+        }
+        pInfo->hasResult = true;
+        //DO_UPDATE_TAG_COLUMNS(pCtx, ts);
+        pResInfo->numOfRes = 1;
+      }
+      break;
+    }
+  } else {  // descending order
+    for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
+      char* data = colDataGetData(pInputCol, i);
+      TSKEY cts = getRowPTs(pInput->pPTS, i);
+      if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf) < cts) {
+        if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
+          pInfo->isNull = true;
+        } else {
+          pInfo->isNull = false;
+          if (IS_VAR_DATA_TYPE(type)) {
+            bytes = varDataTLen(data);
+            pInfo->bytes = bytes;
+          }
+          memcpy(pInfo->buf + sizeof(TSKEY), data, bytes);
+        }
+        *(TSKEY*)(pInfo->buf) = cts;
+        numOfElems++;
+        //handle selectivity
+        if (pCtx->subsidiaries.num > 0) {
+          STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
+          if (!pInfo->hasResult) {
+            saveTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          } else {
+            copyTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          }
+        }
+        pInfo->hasResult = true;
+        pResInfo->numOfRes = 1;
+        //DO_UPDATE_TAG_COLUMNS(pCtx, ts);
+      }
+      break;
+    }
+  }
+
+  SET_VAL(pResInfo, numOfElems, 1);
+  return TSDB_CODE_SUCCESS;
+}
+
+
+int32_t lastRowFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  int32_t          slotId = pCtx->pExpr->base.resSchema.slotId;
+  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
+
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+
+  SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(pResInfo);
+  colDataAppend(pCol, pBlock->info.rows, pRes->buf + sizeof(TSKEY), pRes->isNull);
+  //handle selectivity
+  STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY));
+  setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows);
+
+  return pResInfo->numOfRes;
+}
+
+
 bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SDiffInfo);
   return true;

tests/system-test/0-others/udfTest.py

@@ -32,7 +32,7 @@ class TDTestCase:
                     buildPath = root[:len(root) - len("/build/bin")]
                     break
         return buildPath
-    
+
     def prepare_udf_so(self):
         selfPath = os.path.dirname(os.path.realpath(__file__))
 
@@ -58,7 +58,7 @@ class TDTestCase:
 
 
     def prepare_data(self):
-        
+
         tdSql.execute("drop database if exists db ")
         tdSql.execute("create database if not exists db  duration 300")
         tdSql.execute("use db")
@@ -68,7 +68,7 @@ class TDTestCase:
         tags (t1 int)
         '''
         )
-        
+
         tdSql.execute(
             '''
             create table t1
@@ -150,7 +150,7 @@ class TDTestCase:
             # create aggregate functions
 
             tdSql.execute("create aggregate function udf2 as '%s' outputtype double bufSize 8;"%self.libudf2)
-            
+
             functions = tdSql.getResult("show functions")
             function_nums = len(functions)
             if function_nums == 2:
@@ -175,14 +175,14 @@ class TDTestCase:
         # create aggregate functions
 
         tdSql.execute("create aggregate function udf2 as '%s' outputtype double bufSize 8;"%self.libudf2)
-        
+
         functions = tdSql.getResult("show functions")
         function_nums = len(functions)
         if function_nums == 2:
             tdLog.info("create two udf functions success ")
-       
+
     def basic_udf_query(self):
-        
+
         # scalar functions
 
         tdSql.execute("use db ")
@@ -256,7 +256,7 @@ class TDTestCase:
         tdSql.checkData(0,1,165.247614504)
         tdSql.checkData(0,2,2551.470164435)
         tdSql.checkData(0,3,2.652476145)
-        
+
         # # bug for crash when query sub table
         tdSql.query("select udf2(c1+100) ,udf2(c6-100) ,udf2(c1*100) ,udf2(c6/100) from ct1")
         tdSql.checkData(0,0,378.215547010)
@@ -281,7 +281,7 @@ class TDTestCase:
         tdSql.error("select udf1(num1) , stddev(num1) from tb;")
         tdSql.error("select udf1(num1) , mode(num1) from tb;")
         tdSql.error("select udf1(num1) , HYPERLOGLOG(num1) from tb;")
-        # stable 
+        # stable
         tdSql.error("select udf1(c1) , count(c1) from stb1;")
         tdSql.error("select udf1(c1) , avg(c1) from stb1;")
         tdSql.error("select udf1(c1) , twa(c1) from stb1;")
@@ -302,23 +302,25 @@ class TDTestCase:
         tdSql.query("select ceil(num1) , min(num1) from tb;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(num1) , first(num1) from tb;")
-        
+
         tdSql.query("select abs(num1) , first(num1) from tb;")
-        
+
         tdSql.query("select udf1(num1) , last(num1) from tb;")
-        
+
         tdSql.query("select round(num1) , last(num1) from tb;")
-        
+
         tdSql.query("select udf1(num1) , top(num1,1) from tb;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(num1) , bottom(num1,1) from tb;")
         tdSql.checkRows(1)
-        tdSql.error("select udf1(num1) , last_row(num1) from tb;")
-       
-        tdSql.error("select round(num1) , last_row(num1) from tb;")
-   
-        
-        # stable 
+        tdSql.query("select udf1(num1) , last_row(num1) from tb;")
+        tdSql.checkRows(1)
+
+        tdSql.query("select round(num1) , last_row(num1) from tb;")
+        tdSql.checkRows(1)
+
+
+        # stable
         tdSql.query("select udf1(c1) , max(c1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select abs(c1) , max(c1) from stb1;")
@@ -328,9 +330,9 @@ class TDTestCase:
         tdSql.query("select floor(c1) , min(c1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(c1) , first(c1) from stb1;")
-        
+
         tdSql.query("select udf1(c1) , last(c1) from stb1;")
-        
+
         tdSql.query("select udf1(c1) , top(c1 ,1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select abs(c1) , top(c1 ,1) from stb1;")
@@ -340,9 +342,11 @@ class TDTestCase:
         tdSql.query("select ceil(c1) , bottom(c1,1) from stb1;")
         tdSql.checkRows(1)
 
-        tdSql.error("select udf1(c1) , last_row(c1) from stb1;")
-        tdSql.error("select ceil(c1) , last_row(c1) from stb1;")
-        
+        tdSql.query("select udf1(c1) , last_row(c1) from stb1;")
+        tdSql.checkRows(1)
+        tdSql.query("select ceil(c1) , last_row(c1) from stb1;")
+        tdSql.checkRows(1)
+
         # regular table with compute functions
 
         tdSql.query("select udf1(num1) , abs(num1) from tb;")
@@ -350,7 +354,7 @@ class TDTestCase:
         tdSql.query("select floor(num1) , abs(num1) from tb;")
         tdSql.checkRows(12)
 
-        # # bug need fix 
+        # # bug need fix
 
         #tdSql.query("select udf1(num1) , csum(num1) from tb;")
         #tdSql.checkRows(9)
@@ -382,8 +386,8 @@ class TDTestCase:
         tdSql.checkData(1,0,88)
         tdSql.checkData(1,1,7)
 
-        # bug fix for crash 
-        # order by udf function result 
+        # bug fix for crash
+        # order by udf function result
         for _ in range(50):
             tdSql.query("select udf2(c1) from stb1 group by 1-udf1(c1)")
             print(tdSql.queryResult)
@@ -401,7 +405,7 @@ class TDTestCase:
         tdSql.checkData(0,1,88)
         tdSql.checkData(0,2,-99.990000000)
         tdSql.checkData(0,3,88)
-        
+
         tdSql.query("select sub1.c1, sub2.c2 from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
         tdSql.checkData(0,0,0)
         tdSql.checkData(0,1,0)
@@ -429,7 +433,7 @@ class TDTestCase:
         tdSql.checkData(0,1,168.819430161)
         tdSql.error("select sub1.c1 , udf2(sub1.c1), sub2.c2 ,udf2(sub2.c2) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
 
-        # udf functions with  group by 
+        # udf functions with  group by
         tdSql.query("select udf1(c1) from ct1 group by c1")
         tdSql.checkRows(10)
         tdSql.query("select udf1(c1) from stb1 group by c1")
@@ -452,7 +456,7 @@ class TDTestCase:
         tdSql.query("select udf2(c1) from stb1 group by floor(c1)")
         tdSql.checkRows(11)
 
-        # udf mix with order by  
+        # udf mix with order by
         tdSql.query("select udf2(c1) from stb1 group by floor(c1) order by udf2(c1)")
         tdSql.checkRows(11)
 
@@ -481,7 +485,7 @@ class TDTestCase:
         tdSql.checkData(0,1,169.661427555)
 
     def try_query_sql(self):
-        udf1_sqls = [ 
+        udf1_sqls = [
         "select num1 , udf1(num1) ,num2 ,udf1(num2),num3 ,udf1(num3),num4 ,udf1(num4) from tb" ,
         "select c1 , udf1(c1) ,c2 ,udf1(c2), c3 ,udf1(c3), c4 ,udf1(c4) from stb1 order by c1" ,
         "select udf1(num1) , max(num1) from tb;" ,
@@ -525,7 +529,7 @@ class TDTestCase:
         "select udf2(c1) from stb1 group by udf1(c1)" ,
         "select udf2(c1) from stb1 group by floor(c1)" ,
         "select udf2(c1) from stb1 group by floor(c1) order by udf2(c1)" ,
-    
+
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
@@ -551,7 +555,7 @@ class TDTestCase:
         for aggregate_sql in udf2_sqls:
             tdSql.error(aggregate_sql)
 
-        # create function without aggregate 
+        # create function without aggregate
 
         tdLog.info(" create function with out aggregate ")
         tdSql.query("drop function udf1 ")
@@ -575,8 +579,8 @@ class TDTestCase:
         tdSql.error(" select test(c1) from stb1 ")
         tdSql.error(" select test(c1,c6), test(c6) from stb1 ")
         tdSql.error(" select test(num1,num2), test(num1) from tb ")
-        
-        
+
+
 
     def loop_kill_udfd(self):
 
@@ -585,7 +589,7 @@ class TDTestCase:
             tdLog.exit("taosd not found!")
         else:
             tdLog.info("taosd found in %s" % buildPath)
-        
+
         cfgPath = buildPath + "/../sim/dnode1/cfg"
         udfdPath = buildPath +'/build/bin/udfd'
 
@@ -596,19 +600,19 @@ class TDTestCase:
             tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
             tdSql.checkData(0,0,169.661427555)
             tdSql.checkData(0,1,169.661427555)
-            # stop udfd cmds 
+            # stop udfd cmds
             get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
             processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
             stop_udfd = " kill -9 %s" % processID
             os.system(stop_udfd)
 
             time.sleep(2)
-            
+
             tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
             tdSql.checkData(0,0,169.661427555)
             tdSql.checkData(0,1,169.661427555)
 
-            # # start udfd  cmds 
+            # # start udfd  cmds
             # start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
             # tdLog.info("start udfd : %s " % start_udfd)
 
@@ -643,11 +647,11 @@ class TDTestCase:
             tdDnodes.stop(1)
             tdDnodes.start(1)
             time.sleep(2)
-            
-            
+
+
     def run(self):  # sourcery skip: extract-duplicate-method, remove-redundant-fstring
-        
-        print(" env is ok for all ") 
+
+        print(" env is ok for all ")
         self.prepare_udf_so()
         self.prepare_data()
         self.create_udf_function()
@@ -659,7 +663,7 @@ class TDTestCase:
         time.sleep(2)
         self.basic_udf_query()
         self.test_function_name()
-        
+
 
     def stop(self):
         tdSql.close()

tests/system-test/0-others/udf_create.py

@@ -34,7 +34,7 @@ class TDTestCase:
                     buildPath = root[:len(root) - len("/build/bin")]
                     break
         return buildPath
-    
+
     def prepare_udf_so(self):
         selfPath = os.path.dirname(os.path.realpath(__file__))
 
@@ -60,7 +60,7 @@ class TDTestCase:
 
 
     def prepare_data(self):
-        
+
         tdSql.execute("drop database if exists db ")
         tdSql.execute("create database if not exists db  duration 300")
         tdSql.execute("use db")
@@ -70,7 +70,7 @@ class TDTestCase:
         tags (t1 int)
         '''
         )
-        
+
         tdSql.execute(
             '''
             create table t1
@@ -152,7 +152,7 @@ class TDTestCase:
             # create aggregate functions
 
             tdSql.execute("create aggregate function udf2 as '%s' outputtype double bufSize 8;"%self.libudf2)
-            
+
             functions = tdSql.getResult("show functions")
             function_nums = len(functions)
             if function_nums == 2:
@@ -177,14 +177,14 @@ class TDTestCase:
         # create aggregate functions
 
         tdSql.execute("create aggregate function udf2 as '%s' outputtype double bufSize 8;"%self.libudf2)
-        
+
         functions = tdSql.getResult("show functions")
         function_nums = len(functions)
         if function_nums == 2:
             tdLog.info("create two udf functions success ")
-       
+
     def basic_udf_query(self):
-        
+
         # scalar functions
 
         tdSql.execute("use db ")
@@ -258,7 +258,7 @@ class TDTestCase:
         tdSql.checkData(0,1,165.247614504)
         tdSql.checkData(0,2,2551.470164435)
         tdSql.checkData(0,3,2.652476145)
-        
+
         # # bug for crash when query sub table
         tdSql.query("select udf2(c1+100) ,udf2(c6-100) ,udf2(c1*100) ,udf2(c6/100) from ct1")
         tdSql.checkData(0,0,378.215547010)
@@ -283,7 +283,7 @@ class TDTestCase:
         tdSql.error("select udf1(num1) , stddev(num1) from tb;")
         tdSql.error("select udf1(num1) , mode(num1) from tb;")
         tdSql.error("select udf1(num1) , HYPERLOGLOG(num1) from tb;")
-        # stable 
+        # stable
         tdSql.error("select udf1(c1) , count(c1) from stb1;")
         tdSql.error("select udf1(c1) , avg(c1) from stb1;")
         tdSql.error("select udf1(c1) , twa(c1) from stb1;")
@@ -304,23 +304,25 @@ class TDTestCase:
         tdSql.query("select ceil(num1) , min(num1) from tb;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(num1) , first(num1) from tb;")
-        
+
         tdSql.query("select abs(num1) , first(num1) from tb;")
-        
+
         tdSql.query("select udf1(num1) , last(num1) from tb;")
-        
+
         tdSql.query("select round(num1) , last(num1) from tb;")
-        
+
         tdSql.query("select udf1(num1) , top(num1,1) from tb;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(num1) , bottom(num1,1) from tb;")
         tdSql.checkRows(1)
-        tdSql.error("select udf1(num1) , last_row(num1) from tb;")
-       
-        tdSql.error("select round(num1) , last_row(num1) from tb;")
-   
-        
-        # stable 
+        tdSql.query("select udf1(num1) , last_row(num1) from tb;")
+        tdSql.checkRows(1)
+
+        tdSql.query("select round(num1) , last_row(num1) from tb;")
+        tdSql.checkRows(1)
+
+
+        # stable
         tdSql.query("select udf1(c1) , max(c1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select abs(c1) , max(c1) from stb1;")
@@ -330,9 +332,9 @@ class TDTestCase:
         tdSql.query("select floor(c1) , min(c1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(c1) , first(c1) from stb1;")
-        
+
         tdSql.query("select udf1(c1) , last(c1) from stb1;")
-        
+
         tdSql.query("select udf1(c1) , top(c1 ,1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select abs(c1) , top(c1 ,1) from stb1;")
@@ -342,9 +344,11 @@ class TDTestCase:
         tdSql.query("select ceil(c1) , bottom(c1,1) from stb1;")
         tdSql.checkRows(1)
 
-        tdSql.error("select udf1(c1) , last_row(c1) from stb1;")
-        tdSql.error("select ceil(c1) , last_row(c1) from stb1;")
-        
+        tdSql.query("select udf1(c1) , last_row(c1) from stb1;")
+        tdSql.checkRows(1)
+        tdSql.query("select ceil(c1) , last_row(c1) from stb1;")
+        tdSql.checkRows(1)
+
         # regular table with compute functions
 
         tdSql.query("select udf1(num1) , abs(num1) from tb;")
@@ -352,7 +356,7 @@ class TDTestCase:
         tdSql.query("select floor(num1) , abs(num1) from tb;")
         tdSql.checkRows(12)
 
-        # # bug need fix 
+        # # bug need fix
 
         #tdSql.query("select udf1(num1) , csum(num1) from tb;")
         #tdSql.checkRows(9)
@@ -384,8 +388,8 @@ class TDTestCase:
         tdSql.checkData(1,0,88)
         tdSql.checkData(1,1,7)
 
-        # bug fix for crash 
-        # order by udf function result 
+        # bug fix for crash
+        # order by udf function result
         for _ in range(50):
             tdSql.query("select udf2(c1) from stb1 group by 1-udf1(c1)")
             print(tdSql.queryResult)
@@ -403,7 +407,7 @@ class TDTestCase:
         tdSql.checkData(0,1,88)
         tdSql.checkData(0,2,-99.990000000)
         tdSql.checkData(0,3,88)
-        
+
         tdSql.query("select sub1.c1, sub2.c2 from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
         tdSql.checkData(0,0,0)
         tdSql.checkData(0,1,0)
@@ -431,7 +435,7 @@ class TDTestCase:
         tdSql.checkData(0,1,168.819430161)
         tdSql.error("select sub1.c1 , udf2(sub1.c1), sub2.c2 ,udf2(sub2.c2) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
 
-        # udf functions with  group by 
+        # udf functions with  group by
         tdSql.query("select udf1(c1) from ct1 group by c1")
         tdSql.checkRows(10)
         tdSql.query("select udf1(c1) from stb1 group by c1")
@@ -454,7 +458,7 @@ class TDTestCase:
         tdSql.query("select udf2(c1) from stb1 group by floor(c1)")
         tdSql.checkRows(11)
 
-        # udf mix with order by  
+        # udf mix with order by
         tdSql.query("select udf2(c1) from stb1 group by floor(c1) order by udf2(c1)")
         tdSql.checkRows(11)
 
@@ -483,7 +487,7 @@ class TDTestCase:
         tdSql.checkData(0,1,169.661427555)
 
     def try_query_sql(self):
-        udf1_sqls = [ 
+        udf1_sqls = [
         "select num1 , udf1(num1) ,num2 ,udf1(num2),num3 ,udf1(num3),num4 ,udf1(num4) from tb" ,
         "select c1 , udf1(c1) ,c2 ,udf1(c2), c3 ,udf1(c3), c4 ,udf1(c4) from stb1 order by c1" ,
         "select udf1(num1) , max(num1) from tb;" ,
@@ -527,7 +531,7 @@ class TDTestCase:
         "select udf2(c1) from stb1 group by udf1(c1)" ,
         "select udf2(c1) from stb1 group by floor(c1)" ,
         "select udf2(c1) from stb1 group by floor(c1) order by udf2(c1)" ,
-    
+
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
@@ -563,7 +567,7 @@ class TDTestCase:
         for aggregate_sql in udf2_sqls:
             tdSql.error(aggregate_sql)
 
-        # create function without aggregate 
+        # create function without aggregate
 
         tdLog.info(" create function with out aggregate ")
         tdSql.query("drop function udf1 ")
@@ -587,8 +591,8 @@ class TDTestCase:
         tdSql.error(" select test(c1) from stb1 ")
         tdSql.error(" select test(c1,c6), test(c6) from stb1 ")
         tdSql.error(" select test(num1,num2), test(num1) from tb ")
-        
-        
+
+
 
     def loop_kill_udfd(self):
 
@@ -597,7 +601,7 @@ class TDTestCase:
             tdLog.exit("taosd not found!")
         else:
             tdLog.info("taosd found in %s" % buildPath)
-        
+
         cfgPath = buildPath + "/../sim/dnode1/cfg"
         udfdPath = buildPath +'/build/bin/udfd'
 
@@ -608,19 +612,19 @@ class TDTestCase:
             tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
             tdSql.checkData(0,0,169.661427555)
             tdSql.checkData(0,1,169.661427555)
-            # stop udfd cmds 
+            # stop udfd cmds
             get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
             processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
             stop_udfd = " kill -9 %s" % processID
             os.system(stop_udfd)
 
             time.sleep(2)
-            
+
             tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
             tdSql.checkData(0,0,169.661427555)
             tdSql.checkData(0,1,169.661427555)
 
-            # # start udfd  cmds 
+            # # start udfd  cmds
             # start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
             # tdLog.info("start udfd : %s " % start_udfd)
 
@@ -655,17 +659,17 @@ class TDTestCase:
             tdDnodes.stop(1)
             tdDnodes.start(1)
             time.sleep(2)
-            
-            
+
+
     def run(self):  # sourcery skip: extract-duplicate-method, remove-redundant-fstring
-        
-        print(" env is ok for all ") 
+
+        print(" env is ok for all ")
         self.prepare_udf_so()
         self.prepare_data()
         self.create_udf_function()
         self.basic_udf_query()
         self.unexpected_create()
-        
+
 
     def stop(self):
         tdSql.close()

tests/system-test/0-others/udf_restart_taosd.py

@@ -31,7 +31,7 @@ class TDTestCase:
                     buildPath = root[:len(root) - len("/build/bin")]
                     break
         return buildPath
-    
+
     def prepare_udf_so(self):
         selfPath = os.path.dirname(os.path.realpath(__file__))
 
@@ -57,7 +57,7 @@ class TDTestCase:
 
 
     def prepare_data(self):
-        
+
         tdSql.execute("drop database if exists db ")
         tdSql.execute("create database if not exists db  duration 300")
         tdSql.execute("use db")
@@ -67,7 +67,7 @@ class TDTestCase:
         tags (t1 int)
         '''
         )
-        
+
         tdSql.execute(
             '''
             create table t1
@@ -149,7 +149,7 @@ class TDTestCase:
             # create aggregate functions
 
             tdSql.execute("create aggregate function udf2 as '%s' outputtype double bufSize 8;"%self.libudf2)
-            
+
             functions = tdSql.getResult("show functions")
             function_nums = len(functions)
             if function_nums == 2:
@@ -174,14 +174,14 @@ class TDTestCase:
         # create aggregate functions
 
         tdSql.execute("create aggregate function udf2 as '%s' outputtype double bufSize 8;"%self.libudf2)
-        
+
         functions = tdSql.getResult("show functions")
         function_nums = len(functions)
         if function_nums == 2:
             tdLog.info("create two udf functions success ")
-       
+
     def basic_udf_query(self):
-        
+
         # scalar functions
 
         tdSql.execute("use db ")
@@ -255,7 +255,7 @@ class TDTestCase:
         tdSql.checkData(0,1,165.247614504)
         tdSql.checkData(0,2,2551.470164435)
         tdSql.checkData(0,3,2.652476145)
-        
+
         # # bug for crash when query sub table
         tdSql.query("select udf2(c1+100) ,udf2(c6-100) ,udf2(c1*100) ,udf2(c6/100) from ct1")
         tdSql.checkData(0,0,378.215547010)
@@ -280,7 +280,7 @@ class TDTestCase:
         tdSql.error("select udf1(num1) , stddev(num1) from tb;")
         tdSql.error("select udf1(num1) , mode(num1) from tb;")
         tdSql.error("select udf1(num1) , HYPERLOGLOG(num1) from tb;")
-        # stable 
+        # stable
         tdSql.error("select udf1(c1) , count(c1) from stb1;")
         tdSql.error("select udf1(c1) , avg(c1) from stb1;")
         tdSql.error("select udf1(c1) , twa(c1) from stb1;")
@@ -301,23 +301,25 @@ class TDTestCase:
         tdSql.query("select ceil(num1) , min(num1) from tb;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(num1) , first(num1) from tb;")
-        
+
         tdSql.query("select abs(num1) , first(num1) from tb;")
-        
+
         tdSql.query("select udf1(num1) , last(num1) from tb;")
-        
+
         tdSql.query("select round(num1) , last(num1) from tb;")
-        
+
         tdSql.query("select udf1(num1) , top(num1,1) from tb;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(num1) , bottom(num1,1) from tb;")
         tdSql.checkRows(1)
-        tdSql.error("select udf1(num1) , last_row(num1) from tb;")
-       
-        tdSql.error("select round(num1) , last_row(num1) from tb;")
-   
-        
-        # stable 
+        tdSql.query("select udf1(num1) , last_row(num1) from tb;")
+        tdSql.checkRows(1)
+
+        tdSql.query("select round(num1) , last_row(num1) from tb;")
+        tdSql.checkRows(1)
+
+
+        # stable
         tdSql.query("select udf1(c1) , max(c1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select abs(c1) , max(c1) from stb1;")
@@ -327,9 +329,9 @@ class TDTestCase:
         tdSql.query("select floor(c1) , min(c1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select udf1(c1) , first(c1) from stb1;")
-        
+
         tdSql.query("select udf1(c1) , last(c1) from stb1;")
-        
+
         tdSql.query("select udf1(c1) , top(c1 ,1) from stb1;")
         tdSql.checkRows(1)
         tdSql.query("select abs(c1) , top(c1 ,1) from stb1;")
@@ -339,9 +341,11 @@ class TDTestCase:
         tdSql.query("select ceil(c1) , bottom(c1,1) from stb1;")
         tdSql.checkRows(1)
 
-        tdSql.error("select udf1(c1) , last_row(c1) from stb1;")
-        tdSql.error("select ceil(c1) , last_row(c1) from stb1;")
-        
+        tdSql.query("select udf1(c1) , last_row(c1) from stb1;")
+        tdSql.checkRows(1)
+        tdSql.query("select ceil(c1) , last_row(c1) from stb1;")
+        tdSql.checkRows(1)
+
         # regular table with compute functions
 
         tdSql.query("select udf1(num1) , abs(num1) from tb;")
@@ -349,7 +353,7 @@ class TDTestCase:
         tdSql.query("select floor(num1) , abs(num1) from tb;")
         tdSql.checkRows(12)
 
-        # # bug need fix 
+        # # bug need fix
 
         #tdSql.query("select udf1(num1) , csum(num1) from tb;")
         #tdSql.checkRows(9)
@@ -381,8 +385,8 @@ class TDTestCase:
         tdSql.checkData(1,0,88)
         tdSql.checkData(1,1,7)
 
-        # bug fix for crash 
-        # order by udf function result 
+        # bug fix for crash
+        # order by udf function result
         for _ in range(50):
             tdSql.query("select udf2(c1) from stb1 group by 1-udf1(c1)")
             print(tdSql.queryResult)
@@ -400,7 +404,7 @@ class TDTestCase:
         tdSql.checkData(0,1,88)
         tdSql.checkData(0,2,-99.990000000)
         tdSql.checkData(0,3,88)
-        
+
         tdSql.query("select sub1.c1, sub2.c2 from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
         tdSql.checkData(0,0,0)
         tdSql.checkData(0,1,0)
@@ -428,7 +432,7 @@ class TDTestCase:
         tdSql.checkData(0,1,168.819430161)
         tdSql.error("select sub1.c1 , udf2(sub1.c1), sub2.c2 ,udf2(sub2.c2) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
 
-        # udf functions with  group by 
+        # udf functions with  group by
         tdSql.query("select udf1(c1) from ct1 group by c1")
         tdSql.checkRows(10)
         tdSql.query("select udf1(c1) from stb1 group by c1")
@@ -451,7 +455,7 @@ class TDTestCase:
         tdSql.query("select udf2(c1) from stb1 group by floor(c1)")
         tdSql.checkRows(11)
 
-        # udf mix with order by  
+        # udf mix with order by
         tdSql.query("select udf2(c1) from stb1 group by floor(c1) order by udf2(c1)")
         tdSql.checkRows(11)
 
@@ -480,7 +484,7 @@ class TDTestCase:
         tdSql.checkData(0,1,169.661427555)
 
     def try_query_sql(self):
-        udf1_sqls = [ 
+        udf1_sqls = [
         "select num1 , udf1(num1) ,num2 ,udf1(num2),num3 ,udf1(num3),num4 ,udf1(num4) from tb" ,
         "select c1 , udf1(c1) ,c2 ,udf1(c2), c3 ,udf1(c3), c4 ,udf1(c4) from stb1 order by c1" ,
         "select udf1(num1) , max(num1) from tb;" ,
@@ -507,7 +511,7 @@ class TDTestCase:
         "select c1,c2, udf1(c1,c2) from stb1 group by c1,c2" ,
         "select num1,num2,num3,udf1(num1,num2,num3) from tb" ,
         "select c1,c6,udf1(c1,c6) from stb1 order by ts" ,
-        "select abs(udf1(c1,c6,c1,c6)) , abs(ceil(c1)) from stb1 where c1 is not null  order by ts;" 
+        "select abs(udf1(c1,c6,c1,c6)) , abs(ceil(c1)) from stb1 where c1 is not null  order by ts;"
         ]
         udf2_sqls = ["select udf2(sub1.c1), udf2(sub2.c2) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
         "select udf2(c1) from stb1 group by 1-udf1(c1)" ,
@@ -524,7 +528,7 @@ class TDTestCase:
         "select udf2(c1) from stb1 group by udf1(c1)" ,
         "select udf2(c1) from stb1 group by floor(c1)" ,
         "select udf2(c1) from stb1 group by floor(c1) order by udf2(c1)" ,
-    
+
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
         "select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null" ,
@@ -550,7 +554,7 @@ class TDTestCase:
         for aggregate_sql in udf2_sqls:
             tdSql.error(aggregate_sql)
 
-        # create function without aggregate 
+        # create function without aggregate
 
         tdLog.info(" create function with out aggregate ")
         tdSql.query("drop function udf1 ")
@@ -574,8 +578,8 @@ class TDTestCase:
         tdSql.error(" select test(c1) from stb1 ")
         tdSql.error(" select test(c1,c6), test(c6) from stb1 ")
         tdSql.error(" select test(num1,num2), test(num1) from tb ")
-        
-        
+
+
 
     def loop_kill_udfd(self):
 
@@ -584,7 +588,7 @@ class TDTestCase:
             tdLog.exit("taosd not found!")
         else:
             tdLog.info("taosd found in %s" % buildPath)
-        
+
         cfgPath = buildPath + "/../sim/dnode1/cfg"
         udfdPath = buildPath +'/build/bin/udfd'
 
@@ -595,19 +599,19 @@ class TDTestCase:
             tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
             tdSql.checkData(0,0,169.661427555)
             tdSql.checkData(0,1,169.661427555)
-            # stop udfd cmds 
+            # stop udfd cmds
             get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
             processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
             stop_udfd = " kill -9 %s" % processID
             os.system(stop_udfd)
 
             time.sleep(2)
-            
+
             tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
             tdSql.checkData(0,0,169.661427555)
             tdSql.checkData(0,1,169.661427555)
 
-            # # start udfd  cmds 
+            # # start udfd  cmds
             # start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
             # tdLog.info("start udfd : %s " % start_udfd)
 
@@ -640,19 +644,19 @@ class TDTestCase:
             tdDnodes.stop(1)
             tdDnodes.start(1)
             time.sleep(2)
-            
-            
+
+
     def run(self):  # sourcery skip: extract-duplicate-method, remove-redundant-fstring
-        
-        print(" env is ok for all ") 
+
+        print(" env is ok for all ")
         self.prepare_udf_so()
         self.prepare_data()
         self.create_udf_function()
         self.basic_udf_query()
         self.multi_cols_udf()
         self.restart_taosd_query_udf()
-       
-        
+
+
 
     def stop(self):
         tdSql.close()