Add upstream ID updates in DdlController

src/meta/src/rpc/ddl_controller.rs

@@ -1047,6 +1047,11 @@ impl DdlController {
             }
         }
 
+        // update downstream actors' upstream_actor_id and upstream_fragment_id
+        for actor in &mut union_fragment.actors {
+            actor.upstream_actor_id.extend(sink_actor_ids);
+        }
+
         union_fragment
             .upstream_fragment_ids
             .push(upstream_fragment_id);