refactor(meta): refactor how upstream fragment is handled when creating stream job #14510

xxchan · 2024-01-11T08:11:50Z

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

What's changed and what's your intention?

Mainly because when I trying to add source backfill, I found it's not easy to understand how to handle TableJobType correctly.

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-01-11T08:12:01Z

Current dependencies on/for this PR:

main
- PR refactor(meta): refactor how upstream fragment is handled when creating stream job #14510 👈
  - PR feat(meta): align splits with source actor with an upstream #14171
    - PR feat(stream): add kafka backfill executor #14172
      - PR feat: add kafka backfill frontend #14465

This stack of pull requests is managed by Graphite.

xxchan · 2024-01-11T09:13:02Z

src/meta/src/manager/streaming_job.rs

-#[strum_discriminants(name(DdlType))]
-#[strum_discriminants(vis(pub))]


A step back... 🤣

src/meta/src/controller/fragment.rs

xxchan · 2024-01-11T09:27:03Z

src/meta/src/stream/stream_graph/fragment.rs

+                            // handle MV on MV
+
+                            // Build the extra edges between the upstream `Materialize` and the downstream `StreamScan`
+                            // of the new materialized view.
                            let mview_fragment = upstream_root_fragments


I think this can also be source_fragment for MV on backfill-able source later.

Wait, isn't this also the same for CDC table..?

Oh, the only difference is how to get output_indices, we have as_materialize in this branch.

StrikeW · 2024-01-11T11:27:06Z

src/meta/src/controller/fragment.rs

-                f.fragment_type_mask & PbFragmentTypeFlag::Mview as i32 != 0
-            }
+        fragments.retain(|f| {
+            f.fragment_type_mask & PbFragmentTypeFlag::Source as i32 != 0


A normal table also contains a Source fragment which should not filtered as the root fragment. That's why e2e fails I think.

This makes me think what to do for MV on reuseable source..

~~I came up with a solution: search for Source fragment only if there's no Mview.~~

~~It doesn't work for MV with an upstream MV and an upstream source.~~

It should be: For each upstream table id, if it doesn't have a MV fragment, then look for its source fragment.

Actually we are looking for the bottom-most node in the graph. 🤔

Oh, that's why it's called "Root fragment" 🤣

BugenZhao · 2024-01-11T13:31:02Z

src/meta/src/stream/stream_graph/fragment.rs

@@ -668,6 +675,9 @@ impl CompleteStreamFragmentGraph {

                            (mview_id, edge)
                        }
+                        DdlType::Source | DdlType::Table(_) => {


Is this branch reachable for schema change on a sink-into table?

I don't get it. Can you elaborate?

#14417 Oh, it's a TODO now. Never mind.

BugenZhao · 2024-01-12T06:13:09Z

src/meta/src/manager/catalog/fragment.rs

+                // look for Source fragment if there's no MView fragment
+                fragments.insert(table_id, fragment);


This is quite counter-intu😄tive. May add some comments to show that this is for Source job?

Oh, it's already on pub async fn get_upstream_root_fragments.

yezizp2012

LGTM!

…ng stream job (#14510)

xxchan · 2024-12-02T03:37:53Z

src/meta/src/controller/fragment.rs

+                // look for Source fragment if there's no MView fragment
+                _ = fragments.try_insert(fragment.job_id, fragment);


I thought I made a mistake, then I finally noticed it is try_insert 😅

This was referenced Jan 11, 2024

feat(meta): align splits with source actor with an upstream #14171

Closed

feat(stream): add kafka backfill executor #14172

Merged

feat: add kafka backfill frontend #14465

Closed

github-actions bot added the type/chore label Jan 11, 2024

xxchan force-pushed the 01-11-chore_add_some_comments_for_creating_streaming_job branch from 950e10e to 494ead9 Compare January 11, 2024 09:11

xxchan commented Jan 11, 2024

View reviewed changes

src/meta/src/controller/fragment.rs Show resolved Hide resolved

xxchan force-pushed the 01-11-chore_add_some_comments_for_creating_streaming_job branch from 494ead9 to b5c248d Compare January 11, 2024 09:17

xxchan changed the title ~~chore: add some comments for creating streaming job~~ refactor(meta): make stream graph building more reasonable Jan 11, 2024

github-actions bot added type/refactor and removed type/chore labels Jan 11, 2024

xxchan commented Jan 11, 2024

View reviewed changes

xxchan force-pushed the 01-11-chore_add_some_comments_for_creating_streaming_job branch from b5c248d to 9985531 Compare January 11, 2024 09:30

xxchan changed the title ~~refactor(meta): make stream graph building more reasonable~~ refactor(meta): refactor how upstream fragment is handled when creating stream job Jan 11, 2024

xxchan requested review from StrikeW, yezizp2012 and BugenZhao January 11, 2024 09:31

StrikeW reviewed Jan 11, 2024

View reviewed changes

BugenZhao reviewed Jan 11, 2024

View reviewed changes

chore: add some comments for creating streaming job

2aaa7f5

xxchan force-pushed the 01-11-chore_add_some_comments_for_creating_streaming_job branch from 9985531 to 2aaa7f5 Compare January 12, 2024 01:12

fmt

86dd35d

xxchan requested review from StrikeW and BugenZhao January 12, 2024 03:18

BugenZhao approved these changes Jan 12, 2024

View reviewed changes

StrikeW approved these changes Jan 12, 2024

View reviewed changes

yezizp2012 approved these changes Jan 12, 2024

View reviewed changes

xxchan added this pull request to the merge queue Jan 12, 2024

github-merge-queue bot removed this pull request from the merge queue due to failed status checks Jan 12, 2024

xxchan added this pull request to the merge queue Jan 12, 2024

Merged via the queue into main with commit 681c226 Jan 12, 2024
29 of 30 checks passed

xxchan deleted the 01-11-chore_add_some_comments_for_creating_streaming_job branch January 12, 2024 08:06

Little-Wallace pushed a commit that referenced this pull request Jan 20, 2024

refactor(meta): refactor how upstream fragment is handled when creati…

bf901d1

…ng stream job (#14510)

xxchan commented Dec 2, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(meta): refactor how upstream fragment is handled when creating stream job #14510

refactor(meta): refactor how upstream fragment is handled when creating stream job #14510

xxchan commented Jan 11, 2024 •

edited

Loading

xxchan commented Jan 11, 2024

xxchan Jan 11, 2024

xxchan Jan 11, 2024

xxchan Jan 11, 2024

xxchan Jan 11, 2024

StrikeW Jan 11, 2024

xxchan Jan 12, 2024

xxchan Jan 12, 2024 •

edited

Loading

xxchan Jan 12, 2024 •

edited

Loading

BugenZhao Jan 11, 2024

xxchan Jan 12, 2024 •

edited

Loading

BugenZhao Jan 12, 2024

BugenZhao Jan 12, 2024

BugenZhao Jan 12, 2024

yezizp2012 left a comment

xxchan Dec 2, 2024

		#[strum_discriminants(name(DdlType))]
		#[strum_discriminants(vis(pub))]

		// look for Source fragment if there's no MView fragment
		fragments.insert(table_id, fragment);

		// look for Source fragment if there's no MView fragment
		_ = fragments.try_insert(fragment.job_id, fragment);

refactor(meta): refactor how upstream fragment is handled when creating stream job #14510

refactor(meta): refactor how upstream fragment is handled when creating stream job #14510

Conversation

xxchan commented Jan 11, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

xxchan commented Jan 11, 2024

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 Jan 12, 2024 • edited Loading

Choose a reason for hiding this comment

xxchan Jan 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xxchan Jan 12, 2024 • edited Loading

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

yezizp2012 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xxchan commented Jan 11, 2024 •

edited

Loading

xxchan Jan 12, 2024 •

edited

Loading

xxchan Jan 12, 2024 •

edited

Loading

xxchan Jan 12, 2024 •

edited

Loading