From 43711652356f62467ed689afecd7281d1a8992eb Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao@163.com> Date: Wed, 20 Jul 2022 10:03:50 +0800 Subject: [PATCH] feat(stream): add result group id --- source/libs/executor/src/timewindowoperator.c | 109 ++++++++++++++++-- tests/script/tsim/stream/basic1.sim | 107 +++++++++++++++++ tests/script/tsim/stream/sliding.sim | 20 +++- 3 files changed, 220 insertions(+), 16 deletions(-) diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index b0a74c3002..42dfdc603a 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -660,6 +660,62 @@ void printDataBlock(SSDataBlock* pBlock, const char* flag) { taosMemoryFree(pBuf); } +typedef int32_t (*__compare_fn_t)(void* pKey, void* data, int32_t index); + +int32_t binarySearchCom(void* keyList, int num, void* pKey, int order, __compare_fn_t comparefn) { + int firstPos = 0, lastPos = num - 1, midPos = -1; + int numOfRows = 0; + + if (num <= 0) return -1; + if (order == TSDB_ORDER_DESC) { + // find the first position which is smaller or equal than the key + while (1) { + if (comparefn(pKey, keyList, lastPos) >= 0) return lastPos; + if (comparefn(pKey, keyList, firstPos) == 0) return firstPos; + if (comparefn(pKey, keyList, firstPos) < 0) return firstPos - 1; + + numOfRows = lastPos - firstPos + 1; + midPos = (numOfRows >> 1) + firstPos; + + if (comparefn(pKey, keyList, midPos) < 0) { + lastPos = midPos - 1; + } else if (comparefn(pKey, keyList, midPos) > 0) { + firstPos = midPos + 1; + } else { + break; + } + } + + } else { + // find the first position which is bigger or equal than the key + while (1) { + if (comparefn(pKey, keyList, firstPos) <= 0) return firstPos; + if (comparefn(pKey, keyList, lastPos) == 0) return lastPos; + + if (comparefn(pKey, keyList, lastPos) > 0) { + lastPos = lastPos + 1; + if (lastPos >= num) + return -1; + else + return lastPos; + } + + numOfRows = lastPos - firstPos + 1; + midPos = (numOfRows >> 1) + firstPos; + + if (comparefn(pKey, keyList, midPos) < 0) { + lastPos = midPos - 1; + } else if (comparefn(pKey, keyList, midPos) > 0) { + firstPos = midPos + 1; + } else { + break; + } + } + } + + return midPos; +} + typedef int64_t (*__get_value_fn_t)(void* data, int32_t index); int32_t binarySearch(void* keyList, int num, TSKEY key, int order, __get_value_fn_t getValuefn) { @@ -722,14 +778,31 @@ int64_t getReskey(void* data, int32_t index) { return *(int64_t*)pos->key; } +int32_t compareResKey(void* pKey, void* data, int32_t index) { + SArray* res = (SArray*)data; + SResKeyPos* pos = taosArrayGetP(res, index); + SWinRes* pData = (SWinRes*) pKey; + if (pData->ts == *(int64_t*)pos->key) { + if (pData->groupId > pos->groupId) { + return 1; + } else if (pData->groupId < pos->groupId) { + return -1; + } + return 0; + } else if (pData->ts > *(int64_t*)pos->key) { + return 1; + } + return -1; +} + static int32_t saveResult(int64_t ts, int32_t pageId, int32_t offset, uint64_t groupId, SArray* pUpdated) { int32_t size = taosArrayGetSize(pUpdated); - int32_t index = binarySearch(pUpdated, size, ts, TSDB_ORDER_DESC, getReskey); + SWinRes data = {.ts = ts, .groupId = groupId}; + int32_t index = binarySearchCom(pUpdated, size, &data, TSDB_ORDER_DESC, compareResKey); if (index == -1) { index = 0; } else { - TSKEY resTs = getReskey(pUpdated, index); - if (resTs < ts) { + if (compareResKey(&data, pUpdated, index) > 0) { index++; } else { return TSDB_CODE_SUCCESS; @@ -753,10 +826,10 @@ static int32_t saveResultRow(SResultRow* result, uint64_t groupId, SArray* pUpda return saveResult(result->win.skey, result->pageId, result->offset, groupId, pUpdated); } -static void removeResult(SArray* pUpdated, TSKEY key) { +static void removeResult(SArray* pUpdated, SWinRes* pKey) { int32_t size = taosArrayGetSize(pUpdated); - int32_t index = binarySearch(pUpdated, size, key, TSDB_ORDER_DESC, getReskey); - if (index >= 0 && key == getReskey(pUpdated, index)) { + int32_t index = binarySearchCom(pUpdated, size, pKey, TSDB_ORDER_DESC, compareResKey); + if (index >= 0 && 0 == compareResKey(pKey, pUpdated, index)) { taosArrayRemove(pUpdated, index); } } @@ -765,7 +838,7 @@ static void removeResults(SArray* pWins, SArray* pUpdated) { int32_t size = taosArrayGetSize(pWins); for (int32_t i = 0; i < size; i++) { SWinRes* pW = taosArrayGet(pWins, i); - removeResult(pUpdated, pW->ts); + removeResult(pUpdated, pW); } } @@ -775,14 +848,30 @@ int64_t getWinReskey(void* data, int32_t index) { return pos->ts; } +int32_t compareWinRes(void* pKey, void* data, int32_t index) { + SArray* res = (SArray*)data; + SWinRes* pos = taosArrayGetP(res, index); + SResKeyPos* pData = (SResKeyPos*) pKey; + if (*(int64_t*)pData->key == pos->ts) { + if (pData->groupId > pos->groupId) { + return 1; + } else if (pData->groupId < pos->groupId) { + return -1; + } + return 0; + } else if (*(int64_t*)pData->key > pos->ts) { + return 1; + } + return -1; +} + static void removeDeleteResults(SArray* pUpdated, SArray* pDelWins) { int32_t upSize = taosArrayGetSize(pUpdated); int32_t delSize = taosArrayGetSize(pDelWins); for (int32_t i = 0; i < upSize; i++) { SResKeyPos* pResKey = taosArrayGetP(pUpdated, i); - int64_t key = *(int64_t*)pResKey->key; - int32_t index = binarySearch(pDelWins, delSize, key, TSDB_ORDER_DESC, getWinReskey); - if (index >= 0 && key == getWinReskey(pDelWins, index)) { + int32_t index = binarySearchCom(pDelWins, delSize, pResKey, TSDB_ORDER_DESC, compareWinRes); + if (index >= 0 && 0 == compareWinRes(pResKey, pDelWins, index)) { taosArrayRemove(pDelWins, index); } } diff --git a/tests/script/tsim/stream/basic1.sim b/tests/script/tsim/stream/basic1.sim index 2a6d64bcaf..a6f9860831 100644 --- a/tests/script/tsim/stream/basic1.sim +++ b/tests/script/tsim/stream/basic1.sim @@ -462,6 +462,113 @@ if $data25 != 3 then return -1 endi +sql create database test2 vgroups 1 +sql show databases + +sql use test2 +sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int); +sql create table t1 using st tags(1,1,1); +sql create table t2 using st tags(2,2,2); +sql create table t3 using st tags(2,2,2); +sql create table t4 using st tags(2,2,2); +sql create table t5 using st tags(2,2,2); +sql create stream streams2 trigger at_once into streamt as select _wstart, count(*) c1, sum(a) c3,max(b) c4 from st partition by tbname interval(10s) + +sql insert into t1 values(1648791213000,1,1,1,1.0) t2 values(1648791213000,2,2,2,2.0) t3 values(1648791213000,3,3,3,3.0) t4 values(1648791213000,4,4,4,4.0); + +$loop_count = 0 + +loop0: +sleep 300 + +$loop_count = $loop_count + 1 +if $loop_count == 10 then + return -1 +endi + +sql select * from streamt; + +if $rows != 4 then + print =====rows=$rows + goto loop0 +endi + +sql insert into t1 values(1648791213000,5,5,5,5.0) t2 values(1648791213000,6,6,6,6.0) t5 values(1648791213000,7,7,7,7.0); + + +$loop_count = 0 + +loop1: +sleep 300 + +$loop_count = $loop_count + 1 +if $loop_count == 10 then + return -1 +endi + +sql select * from streamt order by c4 desc; + +if $rows != 5 then + print =====rows=$rows + goto loop1 +endi + +# row 0 +if $data01 != 1 then + print =====data01=$data01 + goto loop1 +endi +if $data02 != 7 then + print =====data02=$data02 + goto loop1 +endi + +# row 1 +if $data11 != 1 then + print =====data11=$data11 + goto loop1 +endi + +if $data12 != 6 then + print =====data12=$data12 + goto loop1 +endi + +# row 2 +if $data21 != 1 then + print =====data21=$data21 + goto loop1 +endi + +if $data22 != 5 then + print =====data22=$data22 + goto loop1 +endi + +sql insert into t1 values(1648791213000,8,8,8,8.0); + +$loop_count = 0 + +loop2: +sleep 300 + +$loop_count = $loop_count + 1 +if $loop_count == 10 then + return -1 +endi + +sql select * from streamt order by c4 desc; + +# row 0 +if $data01 != 1 then + print =====data01=$data01 + goto loop2 +endi + +if $data02 != 8 then + print =====data02=$data02 + goto loop2 +endi system sh/exec.sh -n dnode1 -s stop -x SIGINT diff --git a/tests/script/tsim/stream/sliding.sim b/tests/script/tsim/stream/sliding.sim index f34a50de9d..4364b56d44 100644 --- a/tests/script/tsim/stream/sliding.sim +++ b/tests/script/tsim/stream/sliding.sim @@ -366,18 +366,21 @@ if $data32 != 8 then goto loop1 endi +#$loop_all = 0 +#looptest: + sql drop database IF EXISTS test2; sql drop stream IF EXISTS streams21; sql drop stream IF EXISTS streams22; -sql create database test2 vgroups 2; +sql create database test2 vgroups 6; sql use test2; sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams21 trigger at_once into streamt as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); -sql create stream streams22 trigger at_once into streamt2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s); +sql create stream streams21 trigger at_once into streamt as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s, 5s); +sql create stream streams22 trigger at_once into streamt2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s, 5s); sql insert into t1 values(1648791213000,1,1,1,1.0); sql insert into t1 values(1648791223001,2,2,2,1.1); @@ -394,7 +397,7 @@ sql insert into t2 values(1648791213004,4,10,10,4.1); $loop_count = 0 loop2: -sleep 300 +sleep 100 $loop_count = $loop_count + 1 if $loop_count == 10 then @@ -452,7 +455,7 @@ print step 6 $loop_count = 0 loop3: -sleep 300 +# sleep 300 $loop_count = $loop_count + 1 if $loop_count == 10 then @@ -464,7 +467,7 @@ sql select * from streamt2; # row 0 if $data01 != 4 then print =====data01=$data01 - # goto loop3 + goto loop3 endi if $data02 != 10 then @@ -505,4 +508,9 @@ if $data32 != 8 then goto loop3 endi +$loop_all = $loop_all + 1 +print ============loop_all=$loop_all + +#goto looptest + system sh/stop_dnodes.sh \ No newline at end of file -- GitLab