stb_subtable_value=f'concat(concat("{self.stb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(100))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
ctb_subtable_value=f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(100))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
tb_subtable_value=f'concat(concat("{self.tb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(100))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
ifpartition:
partition_elm=f'partition by {partition}{partition_elm_alias}'
self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}',des_table=self.stb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.stb_name}{partition_elm} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="at_once",subtable_value=stb_subtable_value,fill_value=fill_value,fill_history_value=fill_history_value)
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.ctb_name}{partition_elm} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="at_once",subtable_value=ctb_subtable_value,fill_value=fill_value,fill_history_value=fill_history_value)
iffill_value:
if"value"infill_value.lower():
fill_value='VALUE,1,2,3,4,5,6,7,8,9,10,11'
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {self.tb_name}{partition_elm} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="at_once",subtable_value=tb_subtable_value,fill_value=fill_value,fill_history_value=fill_history_value)
self.tdCom.check_query_data(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {tbname}{partition_elm} interval({self.tdCom.dataDict["interval"]}s) order by wstart',sorted=True)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {tbname}{partition_elm} interval({self.tdCom.dataDict["interval"]}s) order by wstart',sorted=True)
ifself.tdCom.subtable:
fortnamein[self.stb_name,self.ctb_name]:
tdSql.query(f'select * from {self.ctb_name}')
ptn_counter=0
forc1_valueintdSql.queryResult:
ifpartition=="c1":
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
elifpartitionisNone:
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
elifpartition=="abs(c1)":
abs_c1_value=abs(c1_value[1])
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
elifpartition=="tbname"andptn_counter==0:
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
ptn_counter+=1
tdSql.checkEqual(tdSql.queryResult[0][0]>0,True)
tdSql.query(f'select * from {self.tb_name}')
ptn_counter=0
forc1_valueintdSql.queryResult:
ifpartition=="c1":
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
elifpartitionisNone:
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
elifpartition=="abs(c1)":
abs_c1_value=abs(c1_value[1])
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
elifpartition=="tbname"andptn_counter==0:
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.fill_stb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart',fill_value=fill_value)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} where `min(c1)` is not Null order by wstart,`min(c1)`',f'select * from (select _wstart AS wstart, {self.tdCom.fill_stb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart) where `min(c1)` is not Null order by wstart,`min(c1)`',fill_value=fill_value)
else:
if"value"infill_value.lower():
fill_value='VALUE,1,2,3,6,7,8,9,10,11'
ifpartition=="tbname":
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.fill_tb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart',fill_value=fill_value)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} where `min(c1)` is not Null order by wstart,`min(c1)`',f'select * from (select _wstart AS wstart, {self.tdCom.fill_tb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart) where `min(c1)` is not Null order by wstart,`min(c1)`',fill_value=fill_value)
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.fill_stb_source_select_str} from {tbname} where ts >= {start_ts.replace("-","+")} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart',fill_value=fill_value)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart,`min(c1)`',f'select * from (select _wstart AS wstart, {self.tdCom.fill_stb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart) where `min(c1)` is not Null order by wstart,`min(c1)`',fill_value=fill_value)
else:
if"value"infill_value.lower():
fill_value='VALUE,1,2,3,6,7,8,9,10,11'
ifpartition=="tbname":
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.fill_tb_source_select_str} from {tbname} where ts >= {start_ts.replace("-","+")} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart',fill_value=fill_value)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart,`min(c1)`',f'select * from (select _wstart AS wstart, {self.tdCom.fill_tb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart) where `min(c1)` is not Null order by wstart,`min(c1)`',fill_value=fill_value)
self.at_once_interval(interval=random.randint(10,15),partition=self.tdCom.stream_case_when_tbname,case_when=f'case when {self.tdCom.stream_case_when_tbname} = tbname then {self.tdCom.partition_tbname_alias} else tbname end')
ctb_subtable_value=f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
tb_subtable_value=f'concat(concat("{self.tb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
else:
ctb_subtable_value=f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", cast(cast({partition_elm_alias} as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
tb_subtable_value=f'concat(concat("{self.tb_name}_{self.tdCom.subtable_prefix}", cast(cast({partition_elm_alias} as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
# create stb/ctb/tb stream
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {self.ctb_name} partition by {partition}{partition_elm_alias} session(ts, {self.tdCom.dataDict["session"]}s)',trigger_mode="at_once",ignore_expired=ignore_expired,ignore_update=ignore_update,subtable_value=ctb_subtable_value,fill_history_value=fill_history_value)
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tb_stream_des_table,source_sql=f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {self.tb_name} partition by {partition}{partition_elm_alias} session(ts, {self.tdCom.dataDict["session"]}s)',trigger_mode="at_once",ignore_expired=ignore_expired,ignore_update=ignore_update,subtable_value=tb_subtable_value,fill_history_value=fill_history_value)
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.tb_output_select_str} from {self.ctb_stream_des_table} order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {self.ctb_name} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s) order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;',sorted=True)
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.tb_output_select_str} from {self.tb_stream_des_table} order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {self.tb_name} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s) order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;')
else:
fortbnamein[self.tb_name]:
iftbname!=self.tb_name:
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {tbname} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s)',sorted=True)
else:
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {tbname} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s)',sorted=True)
ifself.tdCom.disorder:
ifignore_expired:
fortbnamein[self.ctb_name,self.tb_name]:
iftbname!=self.tb_name:
tdSql.query(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}')
self.tdCom.check_query_data(f'select wstart, {self.tdCom.tb_output_select_str} from {self.ctb_stream_des_table} order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {self.ctb_name} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s) order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;',sorted=True)
self.tdCom.check_query_data(f'select wstart, {self.tdCom.tb_output_select_str} from {self.tb_stream_des_table} order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {self.tb_name} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s) order by wstart, `min(c1)`,`max(c2)`,`sum(c3)`;')
else:
fortbnamein[self.tb_name]:
iftbname!=self.tb_name:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {tbname} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s)',sorted=True)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {tbname} partition by {partition} session(ts, {self.tdCom.dataDict["session"]}s)',sorted=True)
tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;')
elifpartition=="abs(c1)":
ifsubtable:
abs_c1_value=abs(c1_value[1])
tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
else:
tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;')
elifpartition=="tbname"andptn_counter==0:
tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
ptn_counter+=1
tdSql.checkEqual(tdSql.queryResult[0][0]>0,True)
tdSql.query(f'select * from {self.tb_name}')
ptn_counter=0
forc1_valueintdSql.queryResult:
ifc1_value[1]isnotNone:
ifpartition=="c1":
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;')
elifpartition=="abs(c1)":
ifsubtable:
abs_c1_value=abs(c1_value[1])
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
else:
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;')
elifpartition=="tbname"andptn_counter==0:
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
ptn_counter+=1
tdSql.checkEqual(tdSql.queryResult[0][0]>0,True)
defrun(self):
self.at_once_session(session=random.randint(10,15),partition=self.tdCom.stream_case_when_tbname,delete=True,case_when=f'case when {self.tdCom.stream_case_when_tbname} = tbname then {self.tdCom.partition_tbname_alias} else tbname end')
ctb_subtable_value=f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
tb_subtable_value=f'concat(concat("{self.tb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
else:
ctb_subtable_value=f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", cast(cast({partition_elm_alias} as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
tb_subtable_value=f'concat(concat("{self.tb_name}_{self.tdCom.subtable_prefix}", cast(cast({partition_elm_alias} as bigint) as varchar(20))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.ctb_name} partition by {partition}{partition_elm_alias} state_window({stream_state_window})',trigger_mode="at_once",subtable_value=ctb_subtable_value,fill_history_value=fill_history_value)
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {self.tb_name} partition by {partition}{partition_elm_alias} state_window({stream_state_window})',trigger_mode="at_once",subtable_value=tb_subtable_value,fill_history_value=fill_history_value)
tdSql.execute(f'insert into {self.ctb_name} (ts, {state_window_col_name}) values ({self.tdCom.date_time}, {state_window_value})')
ifself.tdCom.updateandi%2==0:
tdSql.execute(f'insert into {self.ctb_name} (ts, {state_window_col_name}) values ({self.tdCom.date_time}, {state_window_value})')
ifself.deleteandi%2!=0:
dt=f'cast({self.tdCom.date_time-1} as timestamp)'
tdSql.execute(f'delete from {self.ctb_name} where ts = {dt}')
tdSql.execute(f'insert into {self.tb_name} (ts, {state_window_col_name}) values ({self.tdCom.date_time}, {state_window_value})')
ifself.tdCom.updateandi%2==0:
tdSql.execute(f'insert into {self.tb_name} (ts, {state_window_col_name}) values ({self.tdCom.date_time}, {state_window_value})')
ifself.deleteandi%2!=0:
tdSql.execute(f'delete from {self.tb_name} where ts = {dt}')
self.tdCom.date_time+=1
# for tbname in [self.stb_name, self.ctb_name, self.tb_name]:
fortbnamein[self.ctb_name,self.tb_name]:
iftbname!=self.tb_name:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {tbname} partition by {partition} state_window({state_window_col_name}) order by wstart,{state_window}',sorted=True)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {tbname} partition by {partition} state_window({state_window_col_name}) order by wstart,{state_window}',sorted=True)
self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}',des_table=self.stb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.stb_name} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="max_delay",watermark=watermark_value,max_delay=max_delay_value,fill_value=fill_value)
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.ctb_name} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="max_delay",watermark=watermark_value,max_delay=max_delay_value,fill_value=fill_value)
iffill_value:
if"value"infill_value.lower():
fill_value='VALUE,1,2,3,4,5,6,7,8,9,10,11'
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {self.tb_name} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="max_delay",watermark=watermark_value,max_delay=max_delay_value,fill_value=fill_value)
self.tdCom.check_query_data(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {tbname} interval({self.tdCom.dataDict["interval"]}s)')
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {tbname} interval({self.tdCom.dataDict["interval"]}s)')
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.fill_stb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts}+{self.tdCom.dataDict["interval"]}s+{fill_watermark_value} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value})',fill_value=fill_value)
else:
if"value"infill_value.lower():
fill_value='VALUE,1,2,3,6,7,8,9,10,11'
self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.fill_tb_source_select_str} from {tbname} where ts >= {start_ts} and ts <= {end_ts}+{self.tdCom.dataDict["interval"]}s+{fill_watermark_value} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value})',fill_value=fill_value)
stb_subtable_value=f'concat(concat("{self.stb_name}_{self.subtable_prefix}", cast(cast(abs(cast({subtable} as int)) as bigint) as varchar(100))), "{self.subtable_suffix}")'ifself.subtableelseNone
self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}',des_table=self.tdCom.ext_stb_stream_des_table,subtable_value=stb_subtable_value,source_sql=f'select _wstart AS wstart, {partitial_tb_source_str} from {self.stb_name} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="max_delay",watermark=watermark_value,max_delay=max_delay_value,fill_value=fill_value,fill_history_value=fill_history_value,stb_field_name_value=stb_field_name_value,tag_value=tag_value,use_exist_stb=use_exist_stb)
tdSql.query(f'select {tag_value} from {self.stb_name}')
tag_value_list=tdSql.queryResult
ifnotfill_value:
self.tdCom.check_query_data(f'select {self.tdCom.stb_filter_des_select_elm} from ext_{self.stb_name}{self.tdCom.des_table_suffix} order by ts;',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.stb_name} interval({self.tdCom.dataDict["interval"]}s)',defined_tag_count=defined_tag_count,tag_value_list=tag_value_list,partition=partition)
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {self.ctb_name} session(ts, {self.tdCom.dataDict["session"]}s)',trigger_mode="max_delay",watermark=watermark_value,max_delay=max_delay_value,fill_history_value=fill_history_value)
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tb_stream_des_table,source_sql=f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {self.tb_name} session(ts, {self.tdCom.dataDict["session"]}s)',trigger_mode="max_delay",watermark=watermark_value,max_delay=max_delay_value,fill_history_value=fill_history_value)
self.tdCom.check_stream(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s)',expected_value,max_delay)
else:
self.tdCom.check_stream(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s)',expected_value,max_delay)
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s)')
else:
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s)')
source_sql=f'select _wstart AS wstart, {self.tdCom.partition_by_stb_source_select_str} from {self.stb_name} partition by {partition_by_elm} interval({self.tdCom.dataDict["interval"]}s)'
else:
source_sql=f'select {self.tdCom.stb_filter_des_select_elm} from {self.stb_name} partition by {partition_by_elm}'
self.tdCom.check_query_data(f'select `{colname}` from {self.stb_name}{self.tdCom.des_table_suffix} order by `{colname}`;',f'select {colname} from {self.stb_name} partition by {partition_by_elm} interval({self.tdCom.dataDict["interval"]}s) order by `{colname}`;')
else:
self.tdCom.check_query_data(f'select {self.tdCom.stb_filter_des_select_elm} from {self.stb_name}{self.tdCom.des_table_suffix} order by c1,c2,c3;',f'select {self.tdCom.stb_filter_des_select_elm} from {self.stb_name} partition by {partition_by_elm} order by c1,c2,c3;')
tdSql.query(f'select `{colname}` from {self.stb_name}{self.tdCom.des_table_suffix} order by `{colname}`;')
res1=tdSql.queryResult
tdSql.query(f'select {colname} from {self.stb_name} partition by {partition_by_elm} interval({self.tdCom.dataDict["interval"]}s) order by `{colname}`;')
res2=tdSql.queryResult
tdSql.checkNotEqual(res1,res2)
else:
self.tdCom.check_query_data(f'select {self.tdCom.stb_filter_des_select_elm} from {self.stb_name}{self.tdCom.des_table_suffix} order by c1,c2,c3;',f'select {self.tdCom.stb_filter_des_select_elm} from {self.stb_name} partition by {partition_by_elm} order by c1,c2,c3;')
self.tdCom.check_query_data(f'select `{colname}` from {self.stb_name}{self.tdCom.des_table_suffix} order by `{colname}`;',f'select {colname} from {self.stb_name} partition by {partition_by_elm} interval({self.tdCom.dataDict["interval"]}s) order by `{colname}`;')
else:
self.tdCom.check_query_data(f'select {self.tdCom.stb_filter_des_select_elm} from {self.stb_name}{self.tdCom.des_table_suffix} order by c1,c2,c3;',f'select {self.tdCom.stb_filter_des_select_elm} from {self.stb_name} partition by {partition_by_elm} order by c1,c2,c3;')
stb_subtable_value=f'concat(concat("{self.stb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(100))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
ctb_subtable_value=f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(100))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
tb_subtable_value=f'concat(concat("{self.tb_name}_{self.tdCom.subtable_prefix}", cast(cast(abs(cast({partition_elm_alias} as int)) as bigint) as varchar(100))), "{self.tdCom.subtable_suffix}")'ifself.tdCom.subtableelseNone
ifpartition:
partition_elm=f'partition by {partition}{partition_elm_alias}'
else:
partition_elm=""
self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}',des_table=self.stb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.stb_name}{partition_elm} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="at_once",subtable_value=stb_subtable_value,fill_history_value=fill_history_value)
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.tdCom.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.ctb_name}{partition_elm} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="at_once",subtable_value=ctb_subtable_value,fill_history_value=fill_history_value)
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tdCom.tb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {self.tb_name}{partition_elm} interval({self.tdCom.dataDict["interval"]}s)',trigger_mode="at_once",subtable_value=tb_subtable_value,fill_history_value=fill_history_value)
self.tdCom.check_query_data(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {tbname}{partition_elm} interval({self.tdCom.dataDict["interval"]}s) order by wstart',sorted=True,reverse_check=reverse_check)
else:
self.tdCom.check_query_data(f'select wstart, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {tbname}{partition_elm} interval({self.tdCom.dataDict["interval"]}s) order by wstart',sorted=True,reverse_check=reverse_check)
ifself.tdCom.subtable:
fortnamein[self.stb_name,self.ctb_name]:
tdSql.query(f'select * from {self.ctb_name}')
ptn_counter=0
forc1_valueintdSql.queryResult:
ifpartition=="c1":
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
elifpartitionisNone:
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
elifpartition=="abs(c1)":
abs_c1_value=abs(c1_value[1])
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
elifpartition=="tbname"andptn_counter==0:
tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
ptn_counter+=1
tdSql.checkEqual(tdSql.queryResult[0][0]>0,True)
tdSql.query(f'select * from {self.tb_name}')
ptn_counter=0
forc1_valueintdSql.queryResult:
ifpartition=="c1":
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
elifpartitionisNone:
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
elifpartition=="abs(c1)":
abs_c1_value=abs(c1_value[1])
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
elifpartition=="tbname"andptn_counter==0:
tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
# self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}', des_table=self.stb_stream_des_table, source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.stb_name} session(ts, {self.tdCom.dataDict["session"]}s)', trigger_mode="window_close", watermark=watermark_value)
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {self.ctb_name} session(ts, {self.tdCom.dataDict["session"]}s)',trigger_mode="window_close",watermark=watermark_value,fill_history_value=fill_history_value)
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tb_stream_des_table,source_sql=f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {self.tb_name} session(ts, {self.tdCom.dataDict["session"]}s)',trigger_mode="window_close",watermark=watermark_value,fill_history_value=fill_history_value)
# for tbname in [self.stb_name, self.ctb_name, self.tb_name]:
ifnotfill_history_value:
fortbnamein[self.ctb_name,self.tb_name]:
iftbname!=self.tb_name:
self.tdCom.check_stream(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s) limit {expected_value}',expected_value)
else:
self.tdCom.check_stream(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s) limit {expected_value}',expected_value)
else:
fortbnamein[self.ctb_name,self.tb_name]:
iftbname!=self.tb_name:
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.stb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s) limit {expected_value+1}')
else:
self.tdCom.check_query_data(f'select wstart, wend-{self.tdCom.dataDict["session"]}s, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, _wend AS wend, {self.tdCom.tb_source_select_str} from {tbname} session(ts, {self.tdCom.dataDict["session"]}s) limit {expected_value+1}')
stb_subtable_value=f'concat(concat("{self.stb_name}_{self.subtable_prefix}", cast(cast(abs(cast({subtable} as int)) as bigint) as varchar(100))), "{self.subtable_suffix}")'ifself.subtableelseNone
self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}',des_table=self.tdCom.ext_stb_stream_des_table,source_sql=f'select _wstart AS wstart, {partitial_tb_source_str} from {self.stb_name} session(ts, {self.tdCom.dataDict["session"]}s)',trigger_mode="window_close",watermark=watermark_value,subtable_value=stb_subtable_value,fill_history_value=fill_history_value,stb_field_name_value=stb_field_name_value,tag_value=tag_value,use_exist_stb=use_exist_stb)
tdSql.query(f'select {tag_value} from {self.stb_name}')
tag_value_list=tdSql.queryResult
self.tdCom.check_query_data(f'select {self.tdCom.stb_filter_des_select_elm} from ext_{self.stb_name}{self.tdCom.des_table_suffix} order by ts',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.stb_name} session(ts, {self.tdCom.dataDict["session"]}s) order by wstart limit {expected_value};',sorted=True,defined_tag_count=defined_tag_count,tag_value_list=tag_value_list,partition=partition)
self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}',des_table=self.ctb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.ctb_name} state_window({state_window_col_name})',trigger_mode="window_close")
self.tdCom.create_stream(stream_name=f'{self.tb_name}{self.tdCom.stream_suffix}',des_table=self.tb_stream_des_table,source_sql=f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {self.tb_name} state_window({state_window_col_name})',trigger_mode="window_close")
self.tdCom.check_stream(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {tbname} state_window({state_window_col_name}) limit {i}',i)
else:
self.tdCom.check_stream(f'select wstart, {self.tdCom.tb_output_select_str} from {tbname}{self.tdCom.des_table_suffix}',f'select _wstart AS wstart, {self.tdCom.tb_source_select_str} from {tbname} state_window({state_window_col_name}) limit {i}',i)