feat: Experimentally introducing sink into table

[shanicky]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

This PR depends on #12240 and #12495

This PR attempts to introduce the sink into table feature.

Following PR #12240, all newly created tables come with a union operator used to aggregate source/dml and the current sink into table. According to the rfc, we create various structures of sink under these three types.

The specific method is very rudimentary. Currently, when we create a streaming job, we will have an affected_table_change, which describes the target table that needs to replace the table. The generated command carries both add and update semantics at this time. We tried to use the combined mutation in PR #12495 to perform this process atomically, but it seems that it may not be necessary, so we may delete this mutation after testing.

Mainly, we still have many restrictions that need to be solved in the future:

1. We can only create 1 sink to the table at present.
2. Tables with sinks cannot be altered at the moment.

The first case, table with row id as the primary key.

dev=> create table m (v1 int, v2 int, v3 int as v1 + v2)  with (
    connector = 'datagen',
    fields.v1.kind = 'sequence',
    fields.v1.start = '1',
    fields.v1.end = '1000',
    datagen.split.num = '5',
) FORMAT PLAIN ENCODE JSON ;
CREATE_TABLE
dev=> create table t (v1 int);
CREATE_TABLE
dev=> explain create sink s into m as select v1 from t with ( type = 'append-only' , force_append_only = 'true');
                                       QUERY PLAN
-----------------------------------------------------------------------------------------
 StreamProject { exprs: [t.v1, null:Int32, (t.v1 + null:Int32) as $expr1, null:Serial] }
 └─StreamProject { exprs: [t.v1, null:Int32, null:Serial] }
   └─StreamSink { type: append-only, columns: [v1, t._row_id(hidden)] }
     └─StreamTableScan { table: t, columns: [v1, _row_id] }
(4 rows)
dev=> create sink s into m as select v1 from t with ( type = 'append-only' , force_append_only = 'true');
CREATE_SINK

table

sink

The second case, append only table

dev=> create table m (v1 int, v2 int, v3 int as v1 + v2) append only with (
    connector = 'datagen',
    fields.v1.kind = 'sequence',
    fields.v1.start = '1',
    fields.v1.end = '1000',
    datagen.split.num = '5',
) FORMAT PLAIN ENCODE JSON ;
NOTICE:  APPEND ONLY TABLE is currently an experimental feature.
CREATE_TABLE
dev=> create table t (v1 int);
CREATE_TABLE
dev=> explain create sink s into m as select v1 from t with ( type = 'append-only' , force_append_only = 'true');
                                       QUERY PLAN
-----------------------------------------------------------------------------------------
 StreamProject { exprs: [t.v1, null:Int32, (t.v1 + null:Int32) as $expr1, null:Serial] }
 └─StreamProject { exprs: [t.v1, null:Int32, null:Serial] }
   └─StreamSink { type: append-only, columns: [v1, t._row_id(hidden)] }
     └─StreamTableScan { table: t, columns: [v1, _row_id] }
(4 rows)
dev=> create sink s into m as select v1 from t with ( type = 'append-only' , force_append_only = 'true');
CREATE_SINK

table

sink

The third case,table with primary key

dev=> create table m (v1 int primary key, v2 int) with {
    connector = 'datagen',
    fields.v1.kind = 'sequence',
    fields.v1.start = '1',
    fields.v1.end = '1000',
    datagen.split.num = '5',
) FORMAT PLAIN ENCODE JSON ;
CREATE_TABLE
dev=> create table t (v1 int);
CREATE_TABLE
dev=> explain create sink s into m as select v1 from t;
                                    QUERY PLAN
----------------------------------------------------------------------------------
 StreamProject { exprs: [t.v1, null:Int32] }
 └─StreamSink { type: upsert, columns: [v1, t._row_id(hidden)], pk: [t._row_id] }
   └─StreamTableScan { table: t, columns: [v1, _row_id] }
(3 rows)
dev=> create sink s into m as select v1 from t;
CREATE_SINK

table

sink

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• All checks passed in ./risedev check (or alias, ./risedev c)

[yezizp2012]

Rest of meta part generally LGTM.

[yezizp2012]

Please add some assertion for some frontend work.