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

What's changed and what's your intention?

The type of the sink (append-only/upsert) can be deduced in the frontend, and we can simply pass this type to computing. The deduction requires 3 pieces of information:

1. The append-only attribute derived by the optimizer
2. The append-only attribute specified by the user: CREATE SINK s FROM t WITH (...., format = 'append_only')
3. The force-append-only attribute specified by the user: CREATE SINK s FROM t WITH (...., force_append_only = 'true')

There are 2 * 2 * 2 = 8 combinations of these 3 attributes. The figure below elaborates on the combinations and their results.

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)

Related issue

Closes #7414

Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

• Connector (sources & sinks)
• SQL commands, functions, and operators

Release note

The user will get an error if the sink has format = append_only in with options while it actually cannot be append-only.

=> CREATE TABLE t (v int);   # t is NOT append-only
CREATE_TABLE

=> CREATE SINK s FROM t WITH (
    connector = 'kafka',
    format = 'append_only',
    ......
);
ERROR:  QueryError: Sink error: The sink cannot be append-only. Please add "force_append_only='true'" in WITH options to force the sink to be append-only. Notice that this will cause the sink executor to drop any UPDATE or DELETE message.

The user can fix this by enforcing the sink to be append-only:

=> CREATE TABLE t (v int);   # t is NOT append-only
CREATE_TABLE

=> CREATE SINK s FROM t WITH (
    connector = 'kafka',
    format = 'append_only',
    force_append_only = 'true',
    ......
);
CREATE_SINK