[jira] [Commented] (FLINK-8545) Implement upsert stream table source

["Fabian Hueske (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-8545?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16348675#comment-16348675 ] 

Fabian Hueske commented on FLINK-8545:
--------------------------------------

Yes, many user have been asking for this feature recently.
Great that you're taking the initiative to work on this!

I'd propose to first start with the {{DataStream}} to {{Table}} upsert conversion and work on the {{UpsertTableSource}} interface later because these will need careful API design.

The keys could be defined similar to the time indicators, i.e.,
{code:java}
DataStream[(String, Long, Int)] input = ???
// upsert with key
Table table = tEnv.upsertFromStream(input, 'a, 'b, 'c.key)
// upsert without key -> single row table
Table table = tEnv.upsertFromStream(input, 'a, 'b, 'c){code}
I think it would be good to be able to declare a time attribute that decides whether an upsert is performed or not.
For example this could look like this:
{code:java}
DataStream[(String, Long, Int)] input = ???
Table table = tEnv.upsertFromStream(input, 'a, 'b.rowtime.upsertOrder, 'c.key){code}
 

 

> Implement upsert stream table source 
> -------------------------------------
>
>                 Key: FLINK-8545
>                 URL: https://issues.apache.org/jira/browse/FLINK-8545
>             Project: Flink
>          Issue Type: New Feature
>          Components: Table API & SQL
>            Reporter: Hequn Cheng
>            Assignee: Hequn Cheng
>            Priority: Major
>
> As more and more users are eager for ingesting data with upsert mode in flink sql/table-api, it is valuable to enable table source with upsert mode. I will provide a design doc later and we can have more discussions. Any suggestions are warmly welcomed !
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)