feat(stream): support the change of vtable in stream #34715
+42
−38
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -28,27 +28,27 @@ def test_stream_meta_change_vtable(self): | |
| """ | ||
|
|
||
| tdStream.createSnode() | ||
| tdSql.execute(f"alter all dnodes 'qdebugflag 143';") | ||
| tdSql.execute(f"alter all dnodes 'stdebugflag 135';") | ||
|
|
||
| streams = [] | ||
| streams.append(self.Basic37208()) # [ok] add ctb and drop ctb from stb | ||
| # streams.append(self.Basic0()) # [ok] add ctb and drop ctb from stb | ||
| # streams.append(self.Basic1()) # [ok] drop data source table | ||
|
|
||
| # # streams.append(self.Basic2()) # tag过滤时，修改tag的值，从满足流条件，到不满足流条件; 从不满足流条件，到满足流条件 [fail] | ||
|
|
||
| # # TD-36750 [流计算开发阶段] 虚拟表+删除pre_filter(cbigint >=1)中cbigint列后，应该没有符合条件的数据了，不会触发计算窗口 | ||
| # # TD-38126 pre_filter 在 %%trows 且触发表为虚拟表时不可用 | ||
| # # streams.append(self.Basic3()) # [ok] | ||
|
|
||
| # streams.append(self.Basic4()) # [ok] | ||
| # streams.append(self.Basic5()) # [ok] | ||
|
|
||
| # # TD-37144 [流计算开发阶段] 删除流结果表后继续触发了也没有重建，不符合预期 | ||
| # # streams.append(self.Basic6()) # [fail] | ||
|
|
||
| # streams.append(self.Basic7()) # [ok] | ||
|
|
||
| tdStream.checkAll(streams) | ||
|
|
||
|
|
@@ -70,30 +70,20 @@ def create(self): | |
| tdSql.execute(f"create table {self.db}.ct5 using {self.db}.{self.stbName} (tint, tbigint)tags(5, 5)") | ||
|
|
||
| tdSql.execute(f"create vtable {self.db}.vct1 (cint from {self.db}.ct1.cint) using {self.db}.{self.vstbName} tags(1,1)") | ||
|
|
||
| tdSql.execute(f"use {self.db}") | ||
|
|
||
| tdSql.execute( | ||
| f"create stream {self.db}.s1 count_window(1) from {self.vstbName} partition by tbname, tint into {self.db}.res_stb OUTPUT_SUBTABLE(CONCAT('res_stb_', tbname)) (firstts, lastts, cnt_v, sum_v, avg_v) as select first(_c0), last_row(_c0), count(cint), sum(cint), avg(cint) from %%trows;" | ||
| ) | ||
|
|
||
| tdSql.execute( | ||
| f"create stream {self.db}.s2 count_window(1) from {self.vstbName} partition by tbname, tint stream_options(DELETE_OUTPUT_TABLE) into {self.db}.res_stb_1 OUTPUT_SUBTABLE(CONCAT('res_stb_1_', tbname)) (firstts, lastts, cnt_v, sum_v, avg_v) as select first(_c0), last_row(_c0), count(cint), sum(cint), avg(cint) from %%trows;" | ||
| ) | ||
|
|
||
| tdSql.execute( | ||
| f"create stream {self.db}.s3 count_window(1) from {self.vstbName} partition by tbname, tint stream_options(PRE_FILTER(tbigint <= 2)) into {self.db}.res_stb_2 OUTPUT_SUBTABLE(CONCAT('res_stb_2_', tbname)) (firstts, lastts, cnt_v, sum_v, avg_v) as select first(_c0), last_row(_c0), count(cint), sum(cint), avg(cint) from {self.vstbName};" | ||
| ) | ||
|
|
||
| def insert1(self): | ||
| tdSql.execute(f"create vtable {self.db}.vct2 (cint from {self.db}.ct2.cint) using {self.db}.{self.vstbName} tags(2,2)") | ||
|
|
@@ -106,13 +96,13 @@ def insert1(self): | |
|
|
||
| def check1(self): | ||
| tdSql.checkResultsByFunc( | ||
| sql=f'select * from information_schema.ins_tables where db_name="{self.db}" and table_name like "res_stb_1%"', | ||
| func=lambda: tdSql.getRows() == 2, | ||
| ) | ||
|
|
||
| tdSql.checkResultsByFunc( | ||
| sql=f'select * from information_schema.ins_tables where db_name="{self.db}" and table_name like "res_stb_%"', | ||
| func=lambda: tdSql.getRows() == 6, | ||
| ) | ||
|
|
||
| tdSql.checkResultsByFunc( | ||
|
|
@@ -165,19 +155,32 @@ def check2(self): | |
|
|
||
| def insert3(self): | ||
| tdSql.execute(f"create vtable {self.db}.vct3 (cint from {self.db}.ct3.cint) using {self.db}.{self.vstbName} tags(10,10)") | ||
| tdSql.execute(f"create vtable {self.db}.vct4 (cint from {self.db}.ct4.cint) using {self.db}.{self.vstbName} tags(-20,-20)") | ||
| tdSql.execute(f"create vtable {self.db}.vct5 (cint from {self.db}.ct2.cint) using {self.db}.{self.vstbName} tags(20,20)") | ||
| time.sleep(5) | ||
| tdSql.execute(f"insert into ct4 values ('2025-01-01 00:01:00', 1);") | ||
| tdSql.execute(f"insert into ct3 values ('2025-01-01 00:01:00', 1);") | ||
| tdSql.execute(f"insert into ct2 values ('2025-01-05 10:01:00', 1);") | ||
|
|
||
| tdSql.execute(f"alter table {self.db}.vct1 set tag tbigint = 30") | ||
| time.sleep(5) | ||
| tdSql.execute(f"insert into {self.db}.ct1 values ('2025-01-03 00:00:00', 1);") | ||
|
|
||
| tdSql.execute(f"alter table {self.db}.vct3 set tag tbigint = -3") | ||
| tdSql.execute(f"alter table {self.db}.vct4 set tag tbigint = -30") | ||
| tdSql.execute(f"alter table {self.db}.vct5 set tag tbigint = 60") | ||
| time.sleep(5) | ||
| tdSql.execute(f"insert into ct4 values ('2025-01-02 00:01:00', 1);") | ||
| tdSql.execute(f"insert into ct3 values ('2025-01-02 00:01:00', 1);") | ||
|
|
||
| tdSql.execute(f"alter vtable {self.db}.vct4 alter column cint set {self.db}.ct2.cint") | ||
| time.sleep(5) | ||
| tdSql.execute(f"insert into ct2 values ('2025-01-05 10:01:10', 1);") | ||
|
|
||
| tdSql.execute(f"alter vtable {self.db}.vct4 alter column cint set null") | ||
| time.sleep(5) | ||
| tdSql.execute(f"insert into ct2 values ('2025-01-05 10:01:11', 1);") | ||
|
Comment on lines
160
to
+182
Contributor
There was a problem hiding this comment. The |
||
|
|
||
| def check3(self): | ||
| tdSql.checkResultsByFunc( | ||
| sql=f'select * from information_schema.ins_tables where db_name="{self.db}" and table_name like "res_stb_%"', | ||
|
Comment on lines
155
to
186
There was a problem hiding this comment. The |
||
|
|
@@ -218,6 +221,7 @@ def check3(self): | |
| and tdSql.compareData(1, 3, 1) | ||
| and tdSql.compareData(1, 4, 1), | ||
| ) | ||
|
|
||
| class Basic0(StreamCheckItem): | ||
| def __init__(self): | ||
| self.db = "sdb0" | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For consistency with other stream definitions in this file, it's better to use lowercase for SQL keywords like
pre_filter. Also, the trailing semicolon in the f-string is unnecessary and can be removed.