From 860bf88f3124bcc63686a363203202cd420de48d Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 30 Jun 2022 17:20:04 +0800 Subject: [PATCH 1/5] feat(query): add last_row function without cache TD-16927 --- include/libs/function/functionMgt.h | 1 + source/libs/function/inc/builtinsimpl.h | 3 + source/libs/function/src/builtins.c | 11 +-- source/libs/function/src/builtinsimpl.c | 126 ++++++++++++++++++++---- 4 files changed, 116 insertions(+), 25 deletions(-) diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h index f03422672d..582cddaf0f 100644 --- a/include/libs/function/functionMgt.h +++ b/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, FUNCTION_TYPE_MAX, FUNCTION_TYPE_MIN, FUNCTION_TYPE_MODE, diff --git a/source/libs/function/inc/builtinsimpl.h b/source/libs/function/inc/builtinsimpl.h index f7e22cb151..8eafb0703e 100644 --- a/source/libs/function/inc/builtinsimpl.h +++ b/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); diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index 23d7185508..9ea07db41e 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -1947,17 +1947,14 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .finalizeFunc = irateFinalize }, { - .name = "last_row", - .type = FUNCTION_TYPE_LAST_ROW, + .name = "last_rowt", + .type = FUNCTION_TYPE_LAST_ROWT, .classification = FUNC_MGT_AGG_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", diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index a75d5a2ecb..cb0c991389 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/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; @@ -2597,15 +2598,6 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { } memcpy(pInfo->buf, data, bytes); *(TSKEY*)(pInfo->buf + bytes) = cts; - //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; @@ -2629,15 +2621,6 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { } memcpy(pInfo->buf, data, bytes); *(TSKEY*)(pInfo->buf + bytes) = cts; - //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); @@ -2763,6 +2746,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; From 1dd89be98bc42b799061678109f3926b75d7fab4 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 30 Jun 2022 17:39:20 +0800 Subject: [PATCH 2/5] enh(query): add last_row function without cache TD-16927 --- include/libs/function/functionMgt.h | 2 +- source/libs/function/src/builtins.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h index 582cddaf0f..5474315434 100644 --- a/include/libs/function/functionMgt.h +++ b/include/libs/function/functionMgt.h @@ -34,7 +34,7 @@ typedef enum EFunctionType { FUNCTION_TYPE_ELAPSED, FUNCTION_TYPE_IRATE, FUNCTION_TYPE_LAST_ROW, - FUNCTION_TYPE_LAST_ROWT, + FUNCTION_TYPE_LAST_ROWT, //TODO: removed FUNCTION_TYPE_MAX, FUNCTION_TYPE_MIN, FUNCTION_TYPE_MODE, diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index bc3b0f165e..b3a627271f 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -1952,7 +1952,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .finalizeFunc = irateFinalize }, { - .name = "last_rowt", + .name = "last_row", .type = FUNCTION_TYPE_LAST_ROWT, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .translateFunc = translateFirstLast, From e38e23a3ecb6e3a9913d1920ca0fa4ab43623e38 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 30 Jun 2022 17:45:55 +0800 Subject: [PATCH 3/5] fix error --- source/libs/function/src/builtinsimpl.c | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 6bff978e69..eb21ee1fb5 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -2598,6 +2598,15 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { } memcpy(pInfo->buf, data, bytes); *(TSKEY*)(pInfo->buf + bytes) = cts; + //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; @@ -2621,6 +2630,15 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { } memcpy(pInfo->buf, data, bytes); *(TSKEY*)(pInfo->buf + bytes) = cts; + //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); From 0692ab4c5e8ec99806af766a45746134b83dcdec Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 30 Jun 2022 17:55:54 +0800 Subject: [PATCH 4/5] enh(query): add last_row selectivity --- source/libs/function/src/builtins.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index b3a627271f..68d89e9ec8 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -1954,7 +1954,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { { .name = "last_row", .type = FUNCTION_TYPE_LAST_ROWT, - .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, + .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .translateFunc = translateFirstLast, .getEnvFunc = getFirstLastFuncEnv, .initFunc = functionSetup, From 61619c940f09d80cb545696007d6de413309ea65 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 30 Jun 2022 19:43:48 +0800 Subject: [PATCH 5/5] fix some test cases --- tests/system-test/0-others/udfTest.py | 92 +++++++++--------- tests/system-test/0-others/udf_create.py | 92 +++++++++--------- .../system-test/0-others/udf_restart_taosd.py | 96 ++++++++++--------- 3 files changed, 146 insertions(+), 134 deletions(-) diff --git a/tests/system-test/0-others/udfTest.py b/tests/system-test/0-others/udfTest.py index 375b7a6272..40f803432a 100644 --- a/tests/system-test/0-others/udfTest.py +++ b/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() diff --git a/tests/system-test/0-others/udf_create.py b/tests/system-test/0-others/udf_create.py index 5f3ab2e863..63650d6edc 100644 --- a/tests/system-test/0-others/udf_create.py +++ b/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() diff --git a/tests/system-test/0-others/udf_restart_taosd.py b/tests/system-test/0-others/udf_restart_taosd.py index 857921e32c..c318980b67 100644 --- a/tests/system-test/0-others/udf_restart_taosd.py +++ b/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()