refactor(rust): Fix topological sort in new streaming engine (pola-rs…

…#18806)

crates/polars-stream/src/execute.rs

@@ -186,6 +186,20 @@ fn run_subgraph(
             for (input, input_pipe) in node.inputs.iter().zip(input_pipes.drain(..)) {
                 if let Some(pipe) = input_pipe {
                     physical_pipes.insert(*input, pipe);
+
+                    // For all the receive ports we just initialized inside spawn(), decrement
+                    // the num_send_ports_not_yet_ready for the node it was connected to and mark
+                    // the node as ready to spawn if all its send ports are connected to
+                    // initialized recv ports.
+                    let sender = graph.pipes[*input].sender;
+                    if let Some(count) = num_send_ports_not_yet_ready.get_mut(sender) {
+                        if *count > 0 {
+                            *count -= 1;
+                            if *count == 0 {
+                                ready.push(sender);
+                            }
+                        }
+                    }
                 }
             }
             for (output, output_pipe) in node.outputs.iter().zip(output_pipes.drain(..)) {
@@ -197,22 +211,6 @@ fn run_subgraph(
             // Reuse the pipe vectors, clearing the borrow it has for next iteration.
             input_pipes = reuse_vec(input_pipes);
             output_pipes = reuse_vec(output_pipes);
-
-            // For all the receive ports we just initialized inside spawn(), decrement
-            // the num_send_ports_not_yet_ready for the node it was connected to and mark
-            // the node as ready to spawn if all its send ports are connected to
-            // initialized recv ports.
-            for input in &node.inputs {
-                let sender = graph.pipes[*input].sender;
-                if let Some(count) = num_send_ports_not_yet_ready.get_mut(sender) {
-                    if *count > 0 {
-                        *count -= 1;
-                        if *count == 0 {
-                            ready.push(sender);
-                        }
-                    }
-                }
-            }
         }
 
         // Spawn tasks for all the physical pipes (no-op on most, but needed for

crates/polars-stream/src/pipe.rs

@@ -95,7 +95,7 @@ impl PhysicalPipe {
     pub fn send_port(&mut self) -> SendPort<'_> {
         assert!(
             matches!(self, Self::SerialReceiver(..) | Self::ParallelReceiver(..)),
-            "PhysicalPipe::send_port must be called on a pipe which only has its send port initialized"
+            "PhysicalPipe::send_port must be called on a pipe which only has its receive port initialized"
         );
         SendPort(self)
     }