feat(stream): support spread、elapsed、histogram、hll

source/libs/function/src/builtinsimpl.c

@@ -142,10 +142,8 @@ typedef struct SElapsedInfo {
 typedef struct SHistoFuncBin {
   double lower;
   double upper;
-  union {
   int64_t count;
   double  percentage;
-  };
 } SHistoFuncBin;
 
 typedef struct SHistoFuncInfo {
@@ -3105,7 +3103,7 @@ int32_t spreadCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
 
   SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
   SSpreadInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
-  spreadTransferInfo(pDBuf, pSBuf);
+  spreadTransferInfo(pSBuf, pDBuf);
   pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);
   return TSDB_CODE_SUCCESS;
 }
@@ -3276,7 +3274,7 @@ int32_t elapsedCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
   SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
   SElapsedInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
 
-  elapsedTransferInfo(pDBuf, pSBuf);
+  elapsedTransferInfo(pSBuf, pDBuf);
   pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);
   return TSDB_CODE_SUCCESS;
 }
@@ -3583,7 +3581,7 @@ int32_t histogramCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
   SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
   SHistoFuncInfo*      pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
 
-  histogramTransferInfo(pDBuf, pSBuf);
+  histogramTransferInfo(pSBuf, pDBuf);
   pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);
   return TSDB_CODE_SUCCESS;
 }
@@ -3779,7 +3777,7 @@ int32_t hllCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
   SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
   SHLLInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
 
-  hllTransferInfo(pDBuf, pSBuf);
+  hllTransferInfo(pSBuf, pDBuf);
   pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);
   return TSDB_CODE_SUCCESS;
 }

tests/script/tsim/stream/session0.sim

@@ -231,8 +231,10 @@ sql use test3;
 sql create table t1(ts timestamp, a int, b int , c int, d double);
 sql create stream streams3 trigger at_once  watermark 1d into streamt3 as select  _wstartts, min(b), a,c from t1 session(ts,10s);
 sql create stream streams4 trigger at_once  watermark 1d into streamt4 as select  _wstartts, max(b), a,c from t1 session(ts,10s);
-sql create stream streams5 trigger at_once  watermark 1d into streamt5 as select  _wstartts, max(b), a,c from t1 session(ts,10s);
-sql create stream streams6 trigger at_once  watermark 1d into streamt6 as select  _wstartts, max(b), a,c from t1 session(ts,10s);
+sql create stream streams5 trigger at_once  watermark 1d into streamt5 as select  _wstartts, top(b,3), a,c from t1 session(ts,10s);
+sql create stream streams6 trigger at_once  watermark 1d into streamt6 as select  _wstartts, bottom(b,3), a,c from t1 session(ts,10s);
+sql create stream streams7 trigger at_once  watermark 1d into streamt7 as select  _wstartts, spread(a), elapsed(ts), hyperloglog(a) from t1 session(ts,10s);
+sql create stream streams8 trigger at_once  watermark 1d into streamt8 as select  _wstartts, histogram(a,"user_input", "[1,3,5,7]", 1), histogram(a,"user_input", "[1,3,5,7]", 0) from t1 session(ts,10s);
 sql insert into t1 values(1648791213001,1,1,1,1.0);
 sql insert into t1 values(1648791213002,2,3,2,3.4);
 sql insert into t1 values(1648791213003,4,9,3,4.8);
@@ -279,4 +281,16 @@ if $rows == 0 then
   goto loop3
 endi
 
+sql select * from streamt7;
+if $rows == 0 then
+  print ======$rows
+  goto loop3
+endi
+
+sql select * from streamt8;
+if $rows == 0 then
+  print ======$rows
+  goto loop3
+endi
+
 system sh/exec.sh -n dnode1 -s stop -x SIGINT