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(sink): Fix sink into table in the recovery of the new version of table graph. #17960

Merged
merged 3 commits into from
Aug 7, 2024
Merged

fix(sink): Fix sink into table in the recovery of the new version of table graph. #17960

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
82adc56
Update `FragmentManager` to handle `NodeBody::Project` and additional…
shanicky Aug 7, 2024
2286647
Refactor `retain_mut` in `FragmentManager`: improve readability, hand…
shanicky Aug 7, 2024
13b9c9d
Remove redundant `&input` refs in `visit_stream_node_cont` methods fo…
shanicky Aug 7, 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
44 changes: 32 additions & 12 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 @@ -704,6 +704,35 @@ impl FragmentManager {

let mut dirty_downstream_table_ids = HashMap::new();

fn union_input_is_clean(
all_fragment_ids: &HashSet<FragmentId>,
input: &StreamNode,
) -> bool {
match &input.node_body {
// for old version sink into table
Some(NodeBody::Merge(merge_node))
if !all_fragment_ids.contains(&merge_node.upstream_fragment_id) =>
{
false
}
// for new version sink into table with project
Some(NodeBody::Project(_)) => {
let merge_stream_node = input.input.iter().exactly_one().expect(
"project of the sink input for the target table should have only one input",
);

if let Some(NodeBody::Merge(merge_node)) = &merge_stream_node.node_body
&& !all_fragment_ids.contains(&merge_node.upstream_fragment_id)
{
false
} else {
true
}
}
_ => true,
}
}

for (table_id, table_fragment) in table_fragments.tree_ref() {
if to_delete_table_ids.contains(table_id) {
continue;
Expand All @@ -722,9 +751,7 @@ impl FragmentManager {
visit_stream_node_cont(actor.nodes.as_ref().unwrap(), |node| {
if let Some(NodeBody::Union(_)) = node.node_body {
for input in &node.input {
if let Some(NodeBody::Merge(merge_node)) = &input.node_body
&& !all_fragment_ids.contains(&merge_node.upstream_fragment_id)
{
if !union_input_is_clean(&all_fragment_ids, input) {
dirty_downstream_table_ids
.insert(*table_id, fragment.fragment_id);
return false;
Expand Down Expand Up @@ -754,15 +781,8 @@ impl FragmentManager {
for actor in &mut fragment.actors {
visit_stream_node_cont_mut(actor.nodes.as_mut().unwrap(), |node| {
if let Some(NodeBody::Union(_)) = node.node_body {
node.input.retain_mut(|input| {
if let Some(NodeBody::Merge(merge_node)) = &mut input.node_body
&& !all_fragment_ids.contains(&merge_node.upstream_fragment_id)
{
false
} else {
true
}
});
node.input
.retain_mut(|input| union_input_is_clean(&all_fragment_ids, input));
}
true
})
Expand Down
Loading