Request feature: sink type = scd type2

[viethqb]

Is your feature request related to a problem? Please describe.

In data analysis scd is an important concept that has applications in many cases. Currently risingwave supports two main sink types: append and upsert. I think sink type = scd type 2 will be a superior feature compared to other applications.
for example mysql cdc => rw => iceberg/doris/starrocks/bigquery (scd type2) => this is awesome.

Describe the solution you'd like

I think the Solution could be: source table must have ID column. sink table has additional columns __rw_valid_flag, __rw_valid_from, __rw_valid_to.

Describe alternatives you've considered

No response

Additional context

No response

[xxchan]

cc @neverchanje @st1page

[xxchan]

I came up with a solution that can work now:

-- your original cdc table
create table t(pk int primary key, v int);

-- transform the cdc table to stream changelog 
-- and add a timestamp column
create table t_changelog(pk int, v int, inserted_at timestamptz as proctime());
create sink s into t_changelog as with c as changelog from t select pk, v from c where changelog_op = 1 or changelog_op = 3 with (type = 'append-only');

-- final scd type 2 sink to your system
CREATE SINK s_scd2 AS 
select *, end_at is null as is_valid from (
    select *, lag(inserted_at) over (partition by pk order by inserted_at desc) as end_at 
    from t_changelog
)
WITH (...);

with sample data:

dev=> insert into t values (1,2);
INSERT 0 1
dev=> insert into t values (1,3);
INSERT 0 1
dev=> insert into t values (1,4);
INSERT 0 1
dev=> insert into t values (2,1);
INSERT 0 1
dev=> select *, end_at is null as is_valid from (
    select *, lag(inserted_at) over (partition by pk order by inserted_at desc) as end_at 
    from t_changelog
);
┌────┬───┬───────────────────────────────┬───────────────────────────────┬──────────┐
│ pk │ v │          inserted_at          │           end_at              │ is_valid │
├────┼───┼───────────────────────────────┼───────────────────────────────┼──────────┤
│  1 │ 4 │ 2024-12-26 03:14:55.993+00:00 │ ∅                             │ t        │
│  1 │ 3 │ 2024-12-26 03:14:54.244+00:00 │ 2024-12-26 03:14:55.993+00:00 │ f        │
│  1 │ 2 │ 2024-12-26 03:14:51.744+00:00 │ 2024-12-26 03:14:54.244+00:00 │ f        │
│  2 │ 1 │ 2024-12-26 03:14:59.244+00:00 │ ∅                             │ t        │
└────┴───┴───────────────────────────────┴───────────────────────────────┴──────────┘
(4 rows)

[viethqb]

@xxchan Thank you so much!