Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: missing update of upstream fragment and actor ids for table replacement #14381

Merged
merged 6 commits into from
Jan 9, 2024
Merged

fix: missing update of upstream fragment and actor ids for table replacement #14381

Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3251fbd
fix: missing update of upstream fragment id for replace table
yezizp2012 Jan 5, 2024
4601cd9
update upstream actors
yezizp2012 Jan 5, 2024
5718950
add intgration test
yezizp2012 Jan 8, 2024
8daca2f
fmt
yezizp2012 Jan 8, 2024
c928406
clippy
yezizp2012 Jan 8, 2024
2d3817a
Merge branch 'main' into fix/update-upstream-fragment-4-replace
yezizp2012 Jan 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 39 additions & 18 deletions src/meta/src/manager/catalog/fragment.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,6 +394,17 @@ impl FragmentManager {

table_fragments.insert(table_id, table_fragment.clone());

// Fragment replace map.
let fragment_replace_map: HashMap<_, _> = merge_updates
.iter()
.map(|update| {
(
update.upstream_fragment_id,
update.new_upstream_fragment_id.unwrap(),
)
})
.collect();

// Update downstream `Merge`s.
let mut merge_updates: HashMap<_, _> = merge_updates
.iter()
Expand All @@ -416,24 +427,34 @@ impl FragmentManager {
.get_mut(table_id)
.with_context(|| format!("table_fragment not exist: id={}", table_id))?;

for actor in table_fragment
.fragments
.values_mut()
.flat_map(|f| &mut f.actors)
{
if let Some(merge_update) = merge_updates.remove(&actor.actor_id) {
assert!(merge_update.removed_upstream_actor_id.is_empty());
assert!(merge_update.new_upstream_fragment_id.is_some());

let stream_node = actor.nodes.as_mut().unwrap();
visit_stream_node(stream_node, |body| {
if let NodeBody::Merge(m) = body
&& m.upstream_fragment_id == merge_update.upstream_fragment_id
{
m.upstream_fragment_id = merge_update.new_upstream_fragment_id.unwrap();
m.upstream_actor_id = merge_update.added_upstream_actor_id.clone();
}
});
for fragment in table_fragment.fragments.values_mut() {
for actor in &mut fragment.actors {
if let Some(merge_update) = merge_updates.remove(&actor.actor_id) {
assert!(merge_update.removed_upstream_actor_id.is_empty());
assert!(merge_update.new_upstream_fragment_id.is_some());

let stream_node = actor.nodes.as_mut().unwrap();
let mut upstream_actor_ids = HashSet::new();
visit_stream_node(stream_node, |body| {
if let NodeBody::Merge(m) = body {
if m.upstream_fragment_id == merge_update.upstream_fragment_id {
m.upstream_fragment_id =
merge_update.new_upstream_fragment_id.unwrap();
m.upstream_actor_id =
merge_update.added_upstream_actor_id.clone();
}
upstream_actor_ids.extend(m.upstream_actor_id.clone());
}
});
actor.upstream_actor_id = upstream_actor_ids.into_iter().collect();
}
}
for upstream_fragment_id in &mut fragment.upstream_fragment_ids {
if let Some(new_upstream_fragment_id) =
fragment_replace_map.get(upstream_fragment_id)
{
*upstream_fragment_id = *new_upstream_fragment_id;
}
}
}
}
Expand Down
Loading