From 860bf88f3124bcc63686a363203202cd420de48d Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 30 Jun 2022 17:20:04 +0800 Subject: [PATCH 1/9] 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/9] 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/9] 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/9] 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/9] 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() From ba265d51b8d5d59267bdfe41157846248224544f Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Fri, 1 Jul 2022 10:29:37 +0800 Subject: [PATCH 6/9] fix: fix stream crash issue --- source/common/src/tdatablock.c | 2 +- source/libs/scalar/src/scalar.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 7b142e3053..2918e44e7b 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -1218,7 +1218,7 @@ int32_t assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src) { for (int32_t i = 0; i < numOfCols; ++i) { SColumnInfoData* pDst = taosArrayGet(dst->pDataBlock, i); SColumnInfoData* pSrc = taosArrayGet(src->pDataBlock, i); - if (pSrc->pData == NULL) { + if (pSrc->pData == NULL && (!IS_VAR_DATA_TYPE(pSrc->info.type))) { continue; } diff --git a/source/libs/scalar/src/scalar.c b/source/libs/scalar/src/scalar.c index cef8ff7075..8e532df3b0 100644 --- a/source/libs/scalar/src/scalar.c +++ b/source/libs/scalar/src/scalar.c @@ -287,7 +287,7 @@ int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarC int32_t code = 0; if (NULL == pParamList) { if (ctx->pBlockList) { - SSDataBlock *pBlock = taosArrayGet(ctx->pBlockList, 0); + SSDataBlock *pBlock = taosArrayGetP(ctx->pBlockList, 0); *rowNum = pBlock->info.rows; } else { *rowNum = 1; From 4e8f3db5964095998813bd9c04a5c59af707432c Mon Sep 17 00:00:00 2001 From: afwerar <1296468573@qq.com> Date: Fri, 1 Jul 2022 11:54:21 +0800 Subject: [PATCH 7/9] os: replace win qsort func --- include/os/osMath.h | 7 ++++ source/common/src/tdatablock.c | 2 +- source/common/src/tdataformat.c | 4 +-- source/libs/executor/src/executil.c | 2 +- source/libs/function/src/taggfunction.c | 4 +-- source/libs/function/src/tpercentile.c | 2 +- source/libs/parser/src/parInsert.c | 8 ++--- source/libs/parser/src/parInsertData.c | 4 +-- source/libs/scalar/src/filter.c | 4 +-- source/os/src/osMath.c | 47 +++++++++++++++++++++++++ source/util/src/tarray.c | 4 +-- source/util/src/tdigest.c | 2 +- source/util/src/terror.c | 2 +- 13 files changed, 73 insertions(+), 19 deletions(-) create mode 100644 source/os/src/osMath.c diff --git a/include/os/osMath.h b/include/os/osMath.h index e13c32422e..b0c75f4dd7 100644 --- a/include/os/osMath.h +++ b/include/os/osMath.h @@ -60,6 +60,13 @@ extern "C" { }) #endif +#ifndef __COMPAR_FN_T +#define __COMPAR_FN_T +typedef int32_t (*__compar_fn_t)(const void *, const void *); +#endif + +void taosSort(void* arr, int64_t sz, int64_t width, __compar_fn_t compar); + #ifdef __cplusplus } #endif diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 7b142e3053..0ec63e90af 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -892,7 +892,7 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo) { int64_t p0 = taosGetTimestampUs(); __compar_fn_t fn = getKeyComparFunc(pColInfoData->info.type, pOrder->order); - qsort(pColInfoData->pData, pDataBlock->info.rows, pColInfoData->info.bytes, fn); + taosSort(pColInfoData->pData, pDataBlock->info.rows, pColInfoData->info.bytes, fn); int64_t p1 = taosGetTimestampUs(); uDebug("blockDataSort easy cost:%" PRId64 ", rows:%d\n", p1 - p0, pDataBlock->info.rows); diff --git a/source/common/src/tdataformat.c b/source/common/src/tdataformat.c index 7c1b31b6e4..83ae442ae7 100644 --- a/source/common/src/tdataformat.c +++ b/source/common/src/tdataformat.c @@ -931,9 +931,9 @@ int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag) { // sort if (isJson) { - qsort(pArray->pData, nTag, sizeof(STagVal), tTagValJsonCmprFn); + taosSort(pArray->pData, nTag, sizeof(STagVal), tTagValJsonCmprFn); } else { - qsort(pArray->pData, nTag, sizeof(STagVal), tTagValCmprFn); + taosSort(pArray->pData, nTag, sizeof(STagVal), tTagValCmprFn); } // get size diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 2561516f6a..415df595e9 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -118,7 +118,7 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) { __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc; - qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn); + taosSort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn); } pGroupResInfo->index = 0; diff --git a/source/libs/function/src/taggfunction.c b/source/libs/function/src/taggfunction.c index 47ae07f823..c8998b9d94 100644 --- a/source/libs/function/src/taggfunction.c +++ b/source/libs/function/src/taggfunction.c @@ -1887,10 +1887,10 @@ static void top_bottom_func_finalizer(SqlFunctionCtx *pCtx) { // user specify the order of output by sort the result according to timestamp if (pCtx->param[1].param.i == PRIMARYKEY_TIMESTAMP_COL_ID) { __compar_fn_t comparator = (pCtx->param[2].param.i == TSDB_ORDER_ASC) ? resAscComparFn : resDescComparFn; - qsort(tvp, (size_t)pResInfo->numOfRes, POINTER_BYTES, comparator); + taosSort(tvp, (size_t)pResInfo->numOfRes, POINTER_BYTES, comparator); } else /*if (pCtx->param[1].param.i > PRIMARYKEY_TIMESTAMP_COL_ID)*/ { __compar_fn_t comparator = (pCtx->param[2].param.i == TSDB_ORDER_ASC) ? resDataAscComparFn : resDataDescComparFn; - qsort(tvp, (size_t)pResInfo->numOfRes, POINTER_BYTES, comparator); + taosSort(tvp, (size_t)pResInfo->numOfRes, POINTER_BYTES, comparator); } GET_TRUE_DATA_TYPE(); diff --git a/source/libs/function/src/tpercentile.c b/source/libs/function/src/tpercentile.c index b40dd8b78d..057d2bc7dc 100644 --- a/source/libs/function/src/tpercentile.c +++ b/source/libs/function/src/tpercentile.c @@ -44,7 +44,7 @@ static SFilePage *loadDataFromFilePage(tMemBucket *pMemBucket, int32_t slotIdx) offset += (int32_t)(pg->num * pMemBucket->bytes); } - qsort(buffer->data, pMemBucket->pSlots[slotIdx].info.size, pMemBucket->bytes, pMemBucket->comparFn); + taosSort(buffer->data, pMemBucket->pSlots[slotIdx].info.size, pMemBucket->bytes, pMemBucket->comparFn); return buffer; } diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c index b97f1df38b..a76640d0b5 100644 --- a/source/libs/parser/src/parInsert.c +++ b/source/libs/parser/src/parInsert.c @@ -790,11 +790,11 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, SParsedDataColInfo* pColIdx[i].schemaColIdx = pColList->boundColumns[i]; pColIdx[i].boundIdx = i; } - qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar); + taosSort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar); for (col_id_t i = 0; i < pColList->numOfBound; ++i) { pColIdx[i].finalIdx = i; } - qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar); + taosSort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar); } if (pColList->numOfCols > pColList->numOfBound) { @@ -2232,11 +2232,11 @@ static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SS pColIdx[i].schemaColIdx = pColList->boundColumns[i]; pColIdx[i].boundIdx = i; } - qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar); + taosSort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar); for (col_id_t i = 0; i < pColList->numOfBound; ++i) { pColIdx[i].finalIdx = i; } - qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar); + taosSort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar); } if (pColList->numOfCols > pColList->numOfBound) { diff --git a/source/libs/parser/src/parInsertData.c b/source/libs/parser/src/parInsertData.c index 56301c072c..84bcef7185 100644 --- a/source/libs/parser/src/parInsertData.c +++ b/source/libs/parser/src/parInsertData.c @@ -295,7 +295,7 @@ void sortRemoveDataBlockDupRowsRaw(STableDataBlocks* dataBuf) { char* pBlockData = pBlocks->data; // todo. qsort is unstable, if timestamp is same, should get the last one - qsort(pBlockData, pBlocks->numOfRows, dataBuf->rowSize, rowDataCompar); + taosSort(pBlockData, pBlocks->numOfRows, dataBuf->rowSize, rowDataCompar); int32_t i = 0; int32_t j = 1; @@ -365,7 +365,7 @@ int sortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKey pBlkKeyTuple = pBlkKeyInfo->pKeyTuple; // todo. qsort is unstable, if timestamp is same, should get the last one - qsort(pBlkKeyTuple, nRows, sizeof(SBlockKeyTuple), rowDataComparStable); + taosSort(pBlkKeyTuple, nRows, sizeof(SBlockKeyTuple), rowDataComparStable); pBlkKeyTuple = pBlkKeyInfo->pKeyTuple; int32_t i = 0; diff --git a/source/libs/scalar/src/filter.c b/source/libs/scalar/src/filter.c index b8bcae161b..9897acfe1a 100644 --- a/source/libs/scalar/src/filter.c +++ b/source/libs/scalar/src/filter.c @@ -2059,7 +2059,7 @@ int32_t filterMergeGroupUnits(SFilterInfo *info, SFilterGroupCtx** gRes, int32_t } if (colIdxi > 1) { - qsort(colIdx, colIdxi, sizeof(uint32_t), getComparFunc(TSDB_DATA_TYPE_USMALLINT, 0)); + taosSort(colIdx, colIdxi, sizeof(uint32_t), getComparFunc(TSDB_DATA_TYPE_USMALLINT, 0)); } for (uint32_t l = 0; l < colIdxi; ++l) { @@ -2294,7 +2294,7 @@ int32_t filterMergeGroups(SFilterInfo *info, SFilterGroupCtx** gRes, int32_t *gR return TSDB_CODE_SUCCESS; } - qsort(gRes, *gResNum, POINTER_BYTES, filterCompareGroupCtx); + taosSort(gRes, *gResNum, POINTER_BYTES, filterCompareGroupCtx); int32_t pEnd = 0, cStart = 0, cEnd = 0; uint32_t pColNum = 0, cColNum = 0; diff --git a/source/os/src/osMath.c b/source/os/src/osMath.c new file mode 100644 index 0000000000..98cd63a831 --- /dev/null +++ b/source/os/src/osMath.c @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#define ALLOW_FORBID_FUNC +#define _DEFAULT_SOURCE +#include "os.h" + +#ifdef WINDOWS +void swapStr(char* j, char* J, int width) { + int i; + char tmp; + for (i = 0; i < width; i++) { + tmp = *j; + *j = *J; + *J = tmp; + j++; + J++; + } +} +#endif + +void taosSort(void* arr, int64_t sz, int64_t width, __compar_fn_t compar) { +#ifdef WINDOWS + int64_t i, j; + for (i = 0; i < sz - 1; i++) { + for (j = 0; j < sz - 1 - i; j++) { + if (compar((char*)arr + j * width, (char*)arr + (j + 1) * width) > 0.00) { + swapStr((char*)arr + j * width, (char*)arr + (j + 1) * width, width); + } + } + } +#else + qsort(arr, sz, width, compar); +#endif +} diff --git a/source/util/src/tarray.c b/source/util/src/tarray.c index ec90e5a9b9..23e79da948 100644 --- a/source/util/src/tarray.c +++ b/source/util/src/tarray.c @@ -373,7 +373,7 @@ void taosArraySort(SArray* pArray, __compar_fn_t compar) { assert(pArray != NULL); assert(compar != NULL); - qsort(pArray->pData, pArray->size, pArray->elemSize, compar); + taosSort(pArray->pData, pArray->size, pArray->elemSize, compar); } void* taosArraySearch(const SArray* pArray, const void* key, __compar_fn_t comparFn, int32_t flags) { @@ -390,7 +390,7 @@ int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t void taosArraySortString(SArray* pArray, __compar_fn_t comparFn) { assert(pArray != NULL); - qsort(pArray->pData, pArray->size, pArray->elemSize, comparFn); + taosSort(pArray->pData, pArray->size, pArray->elemSize, comparFn); } char* taosArraySearchString(const SArray* pArray, const char* key, __compar_fn_t comparFn, int32_t flags) { diff --git a/source/util/src/tdigest.c b/source/util/src/tdigest.c index 56b113fd8f..a722cfeee2 100644 --- a/source/util/src/tdigest.c +++ b/source/util/src/tdigest.c @@ -124,7 +124,7 @@ void tdigestCompress(TDigest *t) { t->num_buffered_pts = 0; t->total_weight += unmerged_weight; - qsort(unmerged_centroids, num_unmerged, sizeof(SCentroid), cmpCentroid); + taosSort(unmerged_centroids, num_unmerged, sizeof(SCentroid), cmpCentroid); memset(&args, 0, sizeof(SMergeArgs)); args.centroids = (SCentroid*)taosMemoryMalloc((size_t)(sizeof(SCentroid) * t->size)); memset(args.centroids, 0, (size_t)(sizeof(SCentroid) * t->size)); diff --git a/source/util/src/terror.c b/source/util/src/terror.c index 3ab307dc3f..58de4970a0 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -614,7 +614,7 @@ static int32_t taosCompareTaosError(const void* a, const void* b) { static TdThreadOnce tsErrorInit = PTHREAD_ONCE_INIT; static void tsSortError(void) { - qsort(errors, sizeof(errors) / sizeof(errors[0]), sizeof(errors[0]), taosCompareTaosError); + taosSort(errors, sizeof(errors) / sizeof(errors[0]), sizeof(errors[0]), taosCompareTaosError); } const char* tstrerror(int32_t err) { From 5adaf57c59edd07118d437b49a289dd17994809c Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Fri, 1 Jul 2022 13:50:49 +0800 Subject: [PATCH 8/9] fix: fix case issue --- .../1-insert/test_stmt_insert_query_ex.py | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/tests/system-test/1-insert/test_stmt_insert_query_ex.py b/tests/system-test/1-insert/test_stmt_insert_query_ex.py index 6d49f6065a..2a66df866d 100644 --- a/tests/system-test/1-insert/test_stmt_insert_query_ex.py +++ b/tests/system-test/1-insert/test_stmt_insert_query_ex.py @@ -184,17 +184,17 @@ class TDTestCase: #query: string Functions - querystmt3=conn.statement("select CHAR_LENGTH(?) from log ") - queryparam3=new_bind_params(1) - print(type(queryparam3)) - queryparam3[0].binary('中文字符') - querystmt3.bind_param(queryparam3) - querystmt3.execute() - result3=querystmt3.use_result() - rows3=result3.fetch_all() - print("4",rows3) - assert rows3[0][0] == 12, 'fourth case is failed' - assert rows3[1][0] == 12, 'fourth case is failed' + querystmt9=conn.statement("select CHAR_LENGTH(?) from log ") + queryparam9=new_bind_params(1) + print(type(queryparam9)) + queryparam9[0].binary('中文字符') + querystmt9.bind_param(queryparam9) + querystmt9.execute() + result9=querystmt9.use_result() + rows9=result9.fetch_all() + print("9",rows9) + assert rows9[0][0] == 12, 'fourth case is failed' + assert rows9[1][0] == 12, 'fourth case is failed' # #query: conversion Functions @@ -259,4 +259,4 @@ class TDTestCase: # add case with filename # tdCases.addWindows(__file__, TDTestCase()) -tdCases.addLinux(__file__, TDTestCase()) \ No newline at end of file +tdCases.addLinux(__file__, TDTestCase()) From f01e5e4b67fa8b7bdc93942cbca27bc550db1f25 Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Fri, 1 Jul 2022 14:37:03 +0800 Subject: [PATCH 9/9] test: modify test case --- tests/system-test/7-tmq/tmqConsFromTsdb.py | 207 ++++++++++++++++++++- 1 file changed, 197 insertions(+), 10 deletions(-) diff --git a/tests/system-test/7-tmq/tmqConsFromTsdb.py b/tests/system-test/7-tmq/tmqConsFromTsdb.py index 8629ab3031..9cd62feee6 100644 --- a/tests/system-test/7-tmq/tmqConsFromTsdb.py +++ b/tests/system-test/7-tmq/tmqConsFromTsdb.py @@ -21,10 +21,10 @@ class TDTestCase: def tmqCase1(self): tdLog.printNoPrefix("======== test case 1: ") - paraDict = {'dbName': 'db1', + paraDict = {'dbName': 'dbt', 'dropFlag': 1, 'event': '', - 'vgroups': 4, + 'vgroups': 1, 'stbName': 'stb', 'colPrefix': 'c', 'tagPrefix': 't', @@ -35,7 +35,7 @@ class TDTestCase: 'rowsPerTbl': 10000, 'batchNum': 10, 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 10, + 'pollDelay': 3, 'showMsg': 1, 'showRow': 1, 'snapshot': 1} @@ -43,7 +43,7 @@ class TDTestCase: topicNameList = ['topic1'] expectRowsList = [] tmqCom.initConsumerTable() - tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=4,replica=1) + tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1) tdLog.info("create stb") tdCom.create_stable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"], column_elm_list=paraDict['colSchema'], tag_elm_list=paraDict['tagSchema']) tdLog.info("create ctb") @@ -52,13 +52,12 @@ class TDTestCase: tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"]) tdDnodes.stop(1) - time.sleep(2) tdDnodes.start(1) tdLog.info("create topics from stb with filter") queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName']) - sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) - # sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + # sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) tdLog.info("create topic sql: %s"%sqlString) tdSql.execute(sqlString) tdSql.query(queryString) @@ -71,31 +70,219 @@ class TDTestCase: topicList = topicNameList[0] ifcheckdata = 1 ifManualCommit = 1 - keyList = 'group.id:cgrp1, enable.auto.commit:false, auto.commit.interval.ms:6000, auto.offset.reset:earliest' + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest' tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) tdLog.info("start consume processor") tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) tdLog.info("wait the consume result") + expectRows = 1 resultList = tmqCom.selectConsumeResult(expectRows) if expectRowsList[0] != resultList[0]: tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[0], resultList[0])) - tdLog.exit("0 tmq consume rows error!") + tdLog.exit("%d tmq consume rows error!"%consumerId) tmqCom.checkFileContent(consumerId, queryString) - time.sleep(10) for i in range(len(topicNameList)): tdSql.query("drop topic %s"%topicNameList[i]) tdLog.printNoPrefix("======== test case 1 end ...... ") + def tmqCase2(self): + tdLog.printNoPrefix("======== test case 2: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbNum': 1, + 'rowsPerTbl': 10000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 3, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 1} + + topicNameList = ['topic1'] + expectRowsList = [] + tmqCom.initConsumerTable() + # tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1) + # tdLog.info("create stb") + # tdCom.create_stable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"], column_elm_list=paraDict['colSchema'], tag_elm_list=paraDict['tagSchema']) + # tdLog.info("create ctb") + # tdCom.create_ctable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"],tag_elm_list=paraDict['tagSchema'],count=paraDict["ctbNum"], default_ctbname_prefix=paraDict['ctbPrefix']) + # tdLog.info("insert data") + # tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"]) + + # tdDnodes.stop(1) + # tdDnodes.start(1) + + tdLog.info("create topics from stb with filter") + queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName']) + # sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + tdSql.query(queryString) + expectRowsList.append(tdSql.getRows()) + totalRowsInserted = expectRowsList[0] + + # init consume info, and start tmq_sim, then check consume result + tdLog.info("insert consume info to consume processor") + consumerId = 1 + expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 7) + topicList = topicNameList[0] + ifcheckdata = 1 + ifManualCommit = 1 + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor 0") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + expectRows = 1 + resultList = tmqCom.selectConsumeResult(expectRows) + + if not (expectrowcnt <= resultList[0] and totalRowsInserted >= resultList[0]): + tdLog.info("act consume rows: %d, expect consume rows between %d and %d"%(resultList[0], expectrowcnt, totalRowsInserted)) + tdLog.exit("%d tmq consume rows error!"%consumerId) + + firstConsumeRows = resultList[0] + + # reinit consume info, and start tmq_sim, then check consume result + tmqCom.initConsumerTable() + consumerId = 2 + expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 3) + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor 1") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + expectRows = 1 + resultList = tmqCom.selectConsumeResult(expectRows) + + actConsumeTotalRows = firstConsumeRows + resultList[0] + + if not (expectrowcnt >= resultList[0] and totalRowsInserted == actConsumeTotalRows): + tdLog.info("act consume rows: %d, expect consume rows <= %d "%(resultList[0], expectrowcnt)) + tdLog.info("and sum of two consume rows: %d , total inserted rows: %d"%(actConsumeTotalRows, totalRowsInserted)) + tdLog.exit("%d tmq consume rows error!"%consumerId) + + time.sleep(10) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + tdLog.printNoPrefix("======== test case 2 end ...... ") + + def tmqCase3(self): + tdLog.printNoPrefix("======== test case 3: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbNum': 10, + 'rowsPerTbl': 10000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 3, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 1} + + topicNameList = ['topic1'] + expectRowsList = [] + tmqCom.initConsumerTable() + # tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1) + # tdLog.info("create stb") + # tdCom.create_stable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"], column_elm_list=paraDict['colSchema'], tag_elm_list=paraDict['tagSchema']) + # tdLog.info("create ctb") + # tdCom.create_ctable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"],tag_elm_list=paraDict['tagSchema'],count=paraDict["ctbNum"], default_ctbname_prefix=paraDict['ctbPrefix']) + # tdLog.info("insert data") + # tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"]) + + # tdDnodes.stop(1) + # tdDnodes.start(1) + + tdLog.info("create topics from stb with filter") + queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName']) + # sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + tdSql.query(queryString) + expectRowsList.append(tdSql.getRows()) + totalRowsInserted = expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] + + # init consume info, and start tmq_sim, then check consume result + tdLog.info("insert consume info to consume processor") + consumerId = 3 + expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 7) + topicList = topicNameList[0] + ifcheckdata = 1 + ifManualCommit = 1 + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor 0") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + expectRows = 1 + resultList = tmqCom.selectConsumeResult(expectRows) + + if not (expectrowcnt <= resultList[0] and totalRowsInserted >= resultList[0]): + tdLog.info("act consume rows: %d, expect consume rows between %d and %d"%(resultList[0], expectrowcnt, totalRowsInserted)) + tdLog.exit("0 tmq consume rows error!") + + firstConsumeRows = resultList[0] + + # reinit consume info, and start tmq_sim, then check consume result + tmqCom.initConsumerTable() + consumerId = 4 + expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 3) + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor 1") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + expectRows = 1 + resultList = tmqCom.selectConsumeResult(expectRows) + + actConsumeTotalRows = firstConsumeRows + resultList[0] + + if not (expectrowcnt >= resultList[0] and totalRowsInserted == actConsumeTotalRows): + tdLog.info("act consume rows: %d, expect consume rows between %d and %d"%(resultList[0], expectrowcnt, totalRowsInserted)) + tdLog.exit("0 tmq consume rows error!") + + time.sleep(10) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + tdLog.printNoPrefix("======== test case 3 end ...... ") + def run(self): tdSql.prepare() self.tmqCase1() + self.tmqCase2() def stop(self): tdSql.close()