feat: add kafka backfill frontend #15602

xxchan · 2024-03-11T08:13:43Z

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

What's changed and what's your intention?

Frontend/meta part of #14172 risingwavelabs/rfcs#72

Tracking issue #16003

How it works

Introduced a session variable enable_shared_source to gate the feature
- TODO: the switch might be problematic. e.g., create source with it disabled, and then create mv with it enabled.

planner:

introduced Logical/StreamSourceBackfill. They are created when planning MV on source.
When to_stream_prost, it becomes merge -> backfill
When CREATE SOURCE, create stream job (with one single node LogicalSource)
TODO: add some planner tests

fragmenter:

added new fragment type SourceBackfill

meta/source manager:

When Backfill fragment is created (and when split change), allocate its splits according to its upstream Source fragment.

meta/build actor graph:

do the job, similar to other special fragments

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.

xxchan · 2024-03-12T09:32:50Z

feat: add kafka backfill frontend #15602 👈
fix: deny create MV on shared CDC source #15635
refactor: introduce Get and WithPropertiesExt #15632
main

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @xxchan and the rest of your teammates on Graphite

src/stream/src/executor/source/source_backfill_executor.rs

fuyufjh

Looks good

proto/catalog.proto

src/connector/src/parser/additional_columns.rs

src/frontend/src/optimizer/plan_node/logical_source_backfill.rs

src/frontend/src/optimizer/plan_node/stream_source_backfill.rs

src/frontend/src/optimizer/plan_node/logical_source_backfill.rs

src/frontend/src/optimizer/plan_node/logical_source.rs

fuyufjh · 2024-03-14T03:38:57Z

proto/catalog.proto

+  // Only used when `has_streaming_job` is `true`.
+  // If `false`, `requires_singleton` will be set in the stream plan.
+  bool is_distributed = 15;
+  reserved "cdc_source_job"; // deprecated


Deprecating cdc_source_job and adding has_streaming_job will break backward compability of existing CDC jobs, right?

Why not renaming cdc_source_job to has_streaming_job?

has_streaming_job is misleading, too. I think we should mention source to indicate what it controls happens when create source, eg. source_has_streaming_job

Changing the field name could be breaking under SQL meta backend. 😐 Not sure why it's not linted.

Why not renaming cdc_source_job to has_streaming_job?

This is indeed the change I made. Number 13 is renamed.

Changing the field name could be breaking under SQL meta backend.

Indeed. 😕 Need to come up with another plan.

do you already have one? Keeping it or introducing another one dedicated to this case both LGTM

New plan:

keep cdc_source_job. New & old CDC sources will both set this field. This field is only used to set require_singleton

Change is_shared to optional (explicit presence). Old CDC sources will have this field set to None. New CDC sources will set to Some(true).

added method is_shared_compatibile to test cdc_source_job first. TBH, this is a little awkward. Better ideas are welcome.

Strictly speaking, it's not necessary to use optional, (cdc_source_job, is_shared) = (true,false) can also work. But I feel this can make it (just slightly) less error-prone.

Old plan:

Rename cdc_source_job to is_shared. This is more elegant, but breaks JSON compatibility for SQL meta backend.

Added is_distributed to set require_singleton in fragmenter for CDC source.

Rename cdc_source_job to is_shared. This is more elegant, but breaks JSON compatibility for SQL meta backend.

Maybe off-topic... I think storing protobuf into JSON sounds bad.

https://protobuf.dev/programming-guides/dos-donts/#text-format-interchange

src/meta/src/stream/source_manager.rs

src/connector/src/parser/additional_columns.rs

src/connector/src/source/kafka/private_link.rs

tabVersion · 2024-03-14T05:18:23Z

proto/catalog.proto

+  // Only used when `has_streaming_job` is `true`.
+  // If `false`, `requires_singleton` will be set in the stream plan.
+  bool is_distributed = 15;
+  reserved "cdc_source_job"; // deprecated


has_streaming_job is misleading, too. I think we should mention source to indicate what it controls happens when create source, eg. source_has_streaming_job

src/frontend/src/optimizer/plan_node/logical_source_backfill.rs

src/frontend/src/optimizer/plan_node/stream_source_backfill.rs

tabVersion · 2024-03-14T06:56:37Z

Had an offline discussion with @xxchan, if we enable the useable source, there will be real actors running and keep fetching data even if no subsquent MVs.

Should have an optim to pause the source in this case to save resources, it is ok to impl it in next pr.

yezizp2012

Rest of meta part LGTM. Can we run single node test in this PR? I'm not sure if it could fail it.

src/meta/src/manager/metadata.rs

src/meta/src/rpc/ddl_controller.rs

src/meta/src/stream/stream_manager.rs

Makefile.toml

BugenZhao · 2024-03-14T08:43:53Z

proto/catalog.proto

+  // Only used when `has_streaming_job` is `true`.
+  // If `false`, `requires_singleton` will be set in the stream plan.
+  bool is_distributed = 15;
+  reserved "cdc_source_job"; // deprecated


Changing the field name could be breaking under SQL meta backend. 😐 Not sure why it's not linted.

BugenZhao · 2024-03-15T07:14:41Z

src/frontend/src/stream_fragmenter/mod.rs

@@ -294,6 +294,7 @@ fn build_fragment(
        }

        NodeBody::StreamCdcScan(_) => {
+            // XXX: Should we use a different flag for CDC scan?


Depending on whether we find the need to distinguish between them.

I think we do need to distinguish them. See the awkward code in tracking_progress_actor_ids. But I'm worried about compatibility so don't want to change it.

src/meta/src/stream/stream_graph/actor.rs

xxchan · 2024-04-01T01:07:32Z

Changes since last review (It should be OK to review commits since last review)

Do not add new logical node. new stream node is enough. Added planner tests
Added basic e2e tests
Changed catalog handling (do not deprecate cdc_source_job) feat: add kafka backfill frontend #15602 (comment)

yezizp2012

Meta part LGTM!

fuyufjh

LGTM

fuyufjh · 2024-04-02T05:57:29Z

src/frontend/src/optimizer/plan_node/stream_source_scan.rs

+        // XXX: do we need to include partition and offset cols here? It's needed by Backfill's input, but maybe not output?
+        // But the source's "schema" contains the hidden columns.


This is by design, so that

users can optionally use these columns in their batch query.

users can optionally use these columns in their streaming query if they specify include in source DDL.

fuyufjh · 2024-04-02T06:11:34Z

src/frontend/src/handler/create_source.rs

@@ -1313,12 +1313,20 @@ pub async fn handle_create_source(
    let sql_pk_names = bind_sql_pk_names(&stmt.columns, &stmt.constraints)?;

    let create_cdc_source_job = with_properties.is_backfillable_cdc_connector();


I saw you renamed is_backfillable_cdc_connector to is_shared_cdc_connector but reverted. 😄 Personally, I will +1 for this renaming to reduce our terminology.

I guess I reverted it by mistake when resolving conflicts 🤡

Wait, it does have a reason for not using "shared". #15635 (comment)

Maybe we can call it "shareable" instead of "backfillable"

src/frontend/src/optimizer/plan_node/logical_source.rs

fuyufjh · 2024-04-02T06:18:58Z

proto/catalog.proto

+  // - Direct CDC sources (mysql & postgresql). Backwards compat note: For old CDC job it's `None`; For new CDC job it's `Some(true)`.
+  // - MQ sources (Kafka, Pulsar, Kinesis, etc.)
+  //
+  // **Should also test `cdc_source_job` for backwards compatibility. Use `is_shared_compatible()` instead of this field directly.**


Looks error-prone to me. I would rather just reuse cdc_source_job even without renaming.

😄 Sure that's acceptable to me

Had better reuse and rename 😄

xxchan marked this pull request as ready for review March 11, 2024 08:13

github-actions bot added the type/feature label Mar 11, 2024

xxchan mentioned this pull request Mar 11, 2024

feat: add kafka backfill frontend #14465

Closed

12 tasks

xxchan force-pushed the xxchan/backfill-fe branch from 228157d to b26f03c Compare March 11, 2024 08:38

xxchan added 3 commits March 12, 2024 17:21

fix: deny create MV on shared CDC source

9c44201

feat: add kafka backfill frontend

55fde18

add config for easy of testing

2285c1f

xxchan force-pushed the xxchan/backfill-fe branch from b26f03c to 2285c1f Compare March 12, 2024 09:32

xxchan changed the base branch from main to 03-12-fix_deny_create_MV_on_shared_CDC_source March 12, 2024 09:32

This was referenced Mar 12, 2024

refactor: introduce Get and WithPropertiesExt #15632

Merged

fix: deny create MV on shared CDC source #15635

Merged

xxchan commented Mar 13, 2024

View reviewed changes

src/stream/src/executor/source/source_backfill_executor.rs Show resolved Hide resolved

cleanup silly issues

6aa8502

xxchan requested review from StrikeW, fuyufjh, tabVersion, yezizp2012, BugenZhao and kwannoel March 13, 2024 02:37

xxchan added 2 commits March 13, 2024 12:08

fix shared cdc source

4c7ae69

fix test

2daa0a8

fuyufjh reviewed Mar 14, 2024

View reviewed changes

tabVersion reviewed Mar 14, 2024

View reviewed changes

yezizp2012 reviewed Mar 14, 2024

View reviewed changes

src/meta/src/manager/metadata.rs Outdated Show resolved Hide resolved

src/meta/src/rpc/ddl_controller.rs Outdated Show resolved Hide resolved

src/meta/src/stream/stream_manager.rs Outdated Show resolved Hide resolved

BugenZhao reviewed Mar 14, 2024

View reviewed changes

Base automatically changed from 03-12-fix_deny_create_MV_on_shared_CDC_source to main March 15, 2024 07:20

BugenZhao reviewed Mar 15, 2024

View reviewed changes

rename has_streaming_job -> is_shared

8d996f4

xxchan mentioned this pull request Mar 28, 2024

improve script for running e2e source tests #15995

Closed

xxchan force-pushed the xxchan/backfill-fe branch from a299d85 to eb64efc Compare March 28, 2024 10:06

refine

17e3779

xxchan force-pushed the xxchan/backfill-fe branch from eb64efc to 17e3779 Compare March 28, 2024 10:07

xxchan mentioned this pull request Mar 28, 2024

Tracking issue for shared source #16003

Closed

21 tasks

xxchan added 3 commits March 29, 2024 00:53

fix

2e5c1c0

Merge remote-tracking branch 'origin/main' into xxchan/backfill-fe

144d7ae

don't deprecate cdc_source_job for compatibility

e174c12

minor fix

98cfb72

xxchan added ci/run-e2e-single-node-tests ci/run-e2e-source-tests ci/run-main-cron and removed ci/run-e2e-source-tests labels Apr 1, 2024

xxchan requested review from StrikeW, fuyufjh, tabVersion, yezizp2012 and BugenZhao April 1, 2024 03:07

xxchan removed the ci/run-main-cron label Apr 1, 2024

yezizp2012 reviewed Apr 1, 2024

View reviewed changes

fuyufjh approved these changes Apr 2, 2024

View reviewed changes

xxchan added 3 commits April 2, 2024 14:46

proto: rename cdc_source_job without renaming

f169dec

rename backfillable -> shareable

aafba45

Merge remote-tracking branch 'origin/main' into xxchan/backfill-fe

a3accdb

xxchan added this pull request to the merge queue Apr 2, 2024

Merged via the queue into main with commit 2afbb6f Apr 2, 2024
30 of 32 checks passed

xxchan deleted the xxchan/backfill-fe branch April 2, 2024 08:39

yezizp2012 mentioned this pull request Apr 2, 2024

feat(sql-backend): change json column type to blob to allow proto field rename #16090

Merged

9 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add kafka backfill frontend #15602

feat: add kafka backfill frontend #15602

xxchan commented Mar 11, 2024 •

edited

Loading

xxchan commented Mar 12, 2024 •

edited

Loading

fuyufjh left a comment

fuyufjh Mar 14, 2024

tabVersion Mar 14, 2024

BugenZhao Mar 14, 2024

xxchan Mar 15, 2024

tabVersion Mar 21, 2024

xxchan Apr 1, 2024

fuyufjh Apr 2, 2024

tabVersion Mar 14, 2024

tabVersion commented Mar 14, 2024

yezizp2012 left a comment

BugenZhao Mar 14, 2024

BugenZhao Mar 15, 2024

xxchan Mar 26, 2024

xxchan commented Apr 1, 2024

yezizp2012 left a comment

fuyufjh left a comment

fuyufjh Apr 2, 2024

fuyufjh Apr 2, 2024

xxchan Apr 2, 2024

xxchan Apr 2, 2024

fuyufjh Apr 2, 2024

xxchan Apr 2, 2024

fuyufjh Apr 2, 2024

		// XXX: do we need to include partition and offset cols here? It's needed by Backfill's input, but maybe not output?
		// But the source's "schema" contains the hidden columns.

		@@ -1313,12 +1313,20 @@ pub async fn handle_create_source(
		let sql_pk_names = bind_sql_pk_names(&stmt.columns, &stmt.constraints)?;

		let create_cdc_source_job = with_properties.is_backfillable_cdc_connector();

feat: add kafka backfill frontend #15602

feat: add kafka backfill frontend #15602

Conversation

xxchan commented Mar 11, 2024 • edited Loading

What's changed and what's your intention?

How it works

Checklist

Documentation

Release note

xxchan commented Mar 12, 2024 • edited Loading

fuyufjh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion commented Mar 14, 2024

yezizp2012 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xxchan commented Apr 1, 2024

yezizp2012 left a comment

Choose a reason for hiding this comment

fuyufjh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xxchan commented Mar 11, 2024 •

edited

Loading

xxchan commented Mar 12, 2024 •

edited

Loading