refactor(rust): Remove unused multiscan code (#22183)

Author: nameexhaustion

crates/polars-stream/src/nodes/io_sinks/csv.rs

@@ -14,7 +14,8 @@ use crate::async_executor::spawn;
 use crate::async_primitives::connector::Receiver;
 use crate::execute::StreamingExecutionState;
 use crate::nodes::io_sinks::parallelize_receive_task;
-use crate::nodes::{JoinHandle, PhaseOutcome, TaskPriority};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
+use crate::nodes::{JoinHandle, TaskPriority};
 
 pub struct CsvSinkNode {
     target: SinkTarget,

crates/polars-stream/src/nodes/io_sinks/ipc.rs

@@ -24,7 +24,8 @@ use crate::async_primitives::connector::{Receiver, connector};
 use crate::async_primitives::distributor_channel::distributor_channel;
 use crate::async_primitives::linearizer::Linearizer;
 use crate::execute::StreamingExecutionState;
-use crate::nodes::{JoinHandle, PhaseOutcome, TaskPriority};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
+use crate::nodes::{JoinHandle, TaskPriority};
 
 pub struct IpcSinkNode {
     target: SinkTarget,

crates/polars-stream/src/nodes/io_sinks/json.rs

@@ -11,7 +11,8 @@ use crate::async_executor::spawn;
 use crate::async_primitives::connector::Receiver;
 use crate::execute::StreamingExecutionState;
 use crate::nodes::io_sinks::parallelize_receive_task;
-use crate::nodes::{JoinHandle, PhaseOutcome, TaskPriority};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
+use crate::nodes::{JoinHandle, TaskPriority};
 
 pub struct NDJsonSinkNode {
     target: SinkTarget,

crates/polars-stream/src/nodes/io_sinks/mod.rs

@@ -8,9 +8,7 @@ use polars_core::prelude::Column;
 use polars_core::schema::SchemaRef;
 use polars_error::PolarsResult;
 
-use super::{
-    ComputeNode, JoinHandle, Morsel, PhaseOutcome, PortState, RecvPort, SendPort, TaskScope,
-};
+use super::{ComputeNode, JoinHandle, Morsel, PortState, RecvPort, SendPort, TaskScope};
 use crate::async_executor::{AbortOnDropHandle, spawn};
 use crate::async_primitives::connector::{Receiver, Sender, connector};
 use crate::async_primitives::distributor_channel;
@@ -19,6 +17,9 @@ use crate::async_primitives::wait_group::WaitGroup;
 use crate::execute::StreamingExecutionState;
 use crate::nodes::TaskPriority;
 
+mod phase;
+use phase::PhaseOutcome;
+
 #[cfg(feature = "csv")]
 pub mod csv;
 #[cfg(feature = "ipc")]

crates/polars-stream/src/nodes/io_sinks/parquet.rs

@@ -27,7 +27,8 @@ use crate::async_primitives::connector::{Receiver, connector};
 use crate::async_primitives::distributor_channel::distributor_channel;
 use crate::async_primitives::linearizer::Linearizer;
 use crate::execute::StreamingExecutionState;
-use crate::nodes::{JoinHandle, PhaseOutcome, TaskPriority};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
+use crate::nodes::{JoinHandle, TaskPriority};
 
 pub struct ParquetSinkNode {
     target: SinkTarget,

crates/polars-stream/src/nodes/io_sinks/partition/by_key.rs

@@ -19,8 +19,9 @@ use crate::async_executor::{AbortOnDropHandle, spawn};
 use crate::execute::StreamingExecutionState;
 use crate::morsel::SourceToken;
 use crate::nodes::io_sinks::partition::{SinkSender, open_new_sink};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
 use crate::nodes::io_sinks::{SinkInputPort, SinkNode, parallelize_receive_task};
-use crate::nodes::{JoinHandle, Morsel, MorselSeq, PhaseOutcome, TaskPriority};
+use crate::nodes::{JoinHandle, Morsel, MorselSeq, TaskPriority};
 
 type Linearized =
     Priority<Reverse<MorselSeq>, (SourceToken, Vec<(Buffer<u8>, Vec<Column>, DataFrame)>)>;

crates/polars-stream/src/nodes/io_sinks/partition/max_size.rs

@@ -18,8 +18,9 @@ use crate::async_primitives::connector::Receiver;
 use crate::async_primitives::distributor_channel::distributor_channel;
 use crate::execute::StreamingExecutionState;
 use crate::nodes::io_sinks::partition::{SinkSender, open_new_sink};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
 use crate::nodes::io_sinks::{SinkInputPort, SinkNode};
-use crate::nodes::{JoinHandle, Morsel, PhaseOutcome, TaskPriority};
+use crate::nodes::{JoinHandle, Morsel, TaskPriority};
 
 pub struct MaxSizePartitionSinkNode {
     input_schema: SchemaRef,

crates/polars-stream/src/nodes/io_sinks/partition/mod.rs

@@ -17,7 +17,8 @@ use crate::async_executor::{AbortOnDropHandle, spawn};
 use crate::async_primitives::wait_group::WaitGroup;
 use crate::async_primitives::{connector, distributor_channel};
 use crate::execute::StreamingExecutionState;
-use crate::nodes::{Morsel, PhaseOutcome, TaskPriority};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
+use crate::nodes::{Morsel, TaskPriority};
 
 pub mod by_key;
 pub mod max_size;

crates/polars-stream/src/nodes/io_sinks/partition/parted.rs

@@ -18,8 +18,9 @@ use crate::async_primitives::connector::Receiver;
 use crate::async_primitives::distributor_channel::distributor_channel;
 use crate::execute::StreamingExecutionState;
 use crate::nodes::io_sinks::partition::{SinkSender, open_new_sink};
+use crate::nodes::io_sinks::phase::PhaseOutcome;
 use crate::nodes::io_sinks::{SinkInputPort, SinkNode};
-use crate::nodes::{JoinHandle, Morsel, PhaseOutcome, TaskPriority};
+use crate::nodes::{JoinHandle, Morsel, TaskPriority};
 
 pub struct PartedPartitionSinkNode {
     // This is not be the same as the input_schema, e.g. when include_key=false then this will not

crates/polars-stream/src/nodes/io_sinks/phase.rs

@@ -0,0 +1,69 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use crate::async_primitives::wait_group::WaitToken;
+
+/// The outcome of a phase in a task.
+///
+/// This indicates whether a task finished (and does not need to be started again) or has stopped
+/// prematurely. When this is dropped without calling `stop`, it is assumed that the task is
+/// finished (most likely because it errored).
+pub struct PhaseOutcome {
+    // This is used to see when phase is finished.
+    #[expect(unused)]
+    consume_token: WaitToken,
+
+    outcome_token: PhaseOutcomeToken,
+}
+
+impl PhaseOutcome {
+    pub fn new_shared_wait(consume_token: WaitToken) -> (PhaseOutcomeToken, Self) {
+        let outcome_token = PhaseOutcomeToken::new();
+        (
+            outcome_token.clone(),
+            Self {
+                consume_token,
+                outcome_token,
+            },
+        )
+    }
+
+    /// Phase ended before the task finished and needs to be called again.
+    pub fn stopped(self) {
+        self.outcome_token.stop();
+    }
+}
+
+/// Token that contains the outcome of a phase.
+///
+/// Namely, this indicates whether a phase finished completely or whether it was stopped before
+/// that.
+#[derive(Clone)]
+pub struct PhaseOutcomeToken {
+    /// - `false` -> finished / panicked
+    /// - `true` -> stopped before finishing
+    stop: Arc<AtomicBool>,
+}
+
+impl PhaseOutcomeToken {
+    pub fn new() -> Self {
+        Self {
+            stop: Arc::new(AtomicBool::new(false)),
+        }
+    }
+
+    /// Indicate that the phase was stopped before finishing.
+    pub fn stop(&self) {
+        self.stop.store(true, Ordering::Relaxed);
+    }
+
+    /// Returns whether the phase was stopped before finishing.
+    pub fn was_stopped(&self) -> bool {
+        self.stop.load(Ordering::Relaxed)
+    }
+
+    /// Returns whether the phase was finished completely.
+    pub fn did_finish(&self) -> bool {
+        !self.was_stopped()
+    }
+}