Merge branch 'main' into gabrielmusat/simplification

Author: gabotechs

src/execution_plans/stage.rs

@@ -262,8 +262,15 @@ impl ExecutionPlan for StageExec {
     ) -> Result<datafusion::execution::SendableRecordBatchStream> {
         let channel_resolver = get_distributed_channel_resolver(context.session_config())?;
 
-        let assigned_stage = self
-            .try_assign_urls(&channel_resolver.get_urls()?)
+        let stage = self
+            .as_any()
+            .downcast_ref::<StageExec>()
+            .expect("Unwrapping myself should always work");
+
+        let urls = channel_resolver.get_urls()?;
+
+        let assigned_stage = stage
+            .try_assign_urls(&urls)
             .map(Arc::new)
             .map_err(|e| DataFusionError::Execution(e.to_string()))?;