how to emit a deletion event for all data in iterating of production logic

[vtygoss <vt...@126.com>]

Hi, Flink community!


I have a problem when iterating the  data production logic.   e.g. 


data production procedure: …-> Table A(change log stream) -> Table B(change log stream) ->Table C(change log stream) ….


production logic of Table B:  insert into table B select * from Table A where a>0; 
new production logic of Table B: insert into table B select * from Table A where a<0; 


So how to emit a deletion event for all data in Table B in order to trigger deletion of old data in Table C? is there a better way for this scenario? 


Thanks for your suggestion!


Best Regards!

Re: how to emit a deletion event for all data in iterating of production logic

[JING ZHANG <be...@gmail.com>]

Hi vtygoss,
You could use `Insert overwrite`[1] syntax if the sink connector supports
`overwrite`.
Otherwise, please try the following 2 solution:
Option A: Define custom source to read the data, and update the header of
`RowData` to `Delete`
Option B: Do a map function which update the header of `RowData` to
`Delete` upon the Datastream read from source

[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/sql/insert/#insert-from-select-queries

Best
JING ZHANG

vtygoss <vt...@126.com> 于2021年8月10日周二 下午7:35写道：

> Hi, Flink community!
>
>
> I have a problem when iterating the  data production logic.   e.g.
>
>
> data production procedure: …-> Table A(change log stream) -> Table
> B(change log stream) ->Table C(change log stream) ….
>
>
> production logic of Table B:  insert into table B select * from Table A
> where a>0;
>
> new production logic of Table B: insert into table B select * from Table A
> where a<0;
>
>
> So how to emit a deletion event for all data in Table B in order to
> trigger deletion of old data in Table C? is there a better way for this
> scenario?
>
>
> Thanks for your suggestion!
>
>
> Best Regards!
>
>
>
>
>
>