feat(cdc): support disable cdc backfill and only consumes from latest changelog

[StrikeW]

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

What's changed and what's your intention?

• Introduces a new snapshot option to allow user to disable cdc backfill and only consumes from the latest changelog.
• Only consume upstream events occurred after the table has been created

The streaming plan is same as table w/ cdc backfill.

create table orders_no_backfill (
   order_id int,
   order_date date,
   customer_name string,
   price decimal,
   product_id int,
   order_status boolean,
   PRIMARY KEY (order_id)
) with (
   snapshot = 'false'
) from pg_source table 'public.orders_tx';

close #14600
related: #13947

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[kwannoel]

Hmm I see that we insert Sam as a record, but where? Do you think it's possible for the cdc tests to be inlined?

Seems a little confusing because we create the table here (in cdc.share_stream.slt).

But we check for the contents in a separate file: (e2e_test/source/cdc/cdc.check_new_rows.slt).

Not sure how they are synced up.

[StrikeW]

Hmm I see that we insert Sam as a record, but where?

In the e2e-source-test.sh which is the "main" of e2e test.

And cdc.check_new_rows.slt is called after cluster restart, which is to test cdc source can resume after recovery. I feel it is also odd to kill the cluster and restart it in the .slt.

[kwannoel]

For CDC Backfill without snapshot, how do we handle the case where CDC updates contain updates/deletes on a non-existent row (i.e. its from the snapshot which was not read).