-- 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 (...);

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)