[jira] [Commented] (FLINK-39261) Add FROM_CHANGELOG built-in process table function (4.1.3.1 upsert stream to upsert stream)

["Timo Walther (Jira)" <ji...@apache.org>]

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

Timo Walther commented on FLINK-39261:
--------------------------------------

Fixed in master: 4997ca40a3f757088be05f117612066cc3382a4f

> Add FROM_CHANGELOG built-in process table function (4.1.3.1 upsert stream to upsert stream)
> -------------------------------------------------------------------------------------------
>
>                 Key: FLINK-39261
>                 URL: https://issues.apache.org/jira/browse/FLINK-39261
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Table SQL / API
>            Reporter: Gustavo de Morais
>            Assignee: Ramin Gharib
>            Priority: Major
>              Labels: pull-request-available
>             Fix For: 2.4.0
>
>
> This is the first implementation of FROM_CHANGELOG. The idea is to implement the use cases one by one interactively. With this ticket we will write the first version which allows flat mapping of an upsert stream.
> Implement FROM_CHANGELOG built-in PTF as specified in FLIP-564, section 4.2.2.2 (retract/upsert stream to upsert stream, flat mode). This PTF converts an append-only stream of changelog records (the input table) into a dynamic table
> h4. [4.1.3.1 Upsert stream to upsert stream|https://cwiki.apache.org/confluence/display/FLINK/FLIP-564%3A+Support+FROM_CHANGELOG+and+TO_CHANGELOG+built-in+PTFs#4131-upsert-stream-to-upsert-stream]
> Converting flat append records with explicit operation codes to a Flink upsert stream.
>  * {*}Input{*}: Flat append records with an operation field (e.g., {{__op}} with values 'c', 'u', 'd').
>  * {*}Output{*}: Table backed by a Flink upsert stream \{+I, +U, -D}.
>  * {*}State{*}: Not required. Updates are directly emitted as UPDATE_AFTER.
>  * {*}Key{*}: A primary key is required on the source table so that we can produce updates for this key.
>  
> {{}}
> {code:java}
> SELECT * FROM FROM_CHANGELOG(
>     input => TABLE cdc_stream PARTITION BY partition_key,
>     uid => 'my-cdc-ptf',
>     op => DESCRIPTOR(__op),
>     op_mapping => MAP[
>         'c, r', 'INSERT',
>         'd', 'DELETE',
>         'u', 'UPDATE_AFTER'
>     ]
> ){code}
> {{}}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)