chore: Simplifying SortPreservingMergeStream to use generators instead of state machine

Cargo.toml

@@ -159,6 +159,7 @@ datafusion-session = { path = "datafusion/session", version = "54.0.0" }
 datafusion-spark = { path = "datafusion/spark", version = "54.0.0" }
 datafusion-sql = { path = "datafusion/sql", version = "54.0.0" }
 datafusion-substrait = { path = "datafusion/substrait", version = "54.0.0" }
+genawaiter = { version = "0.99.1", default-features = false }
 
 doc-comment = "0.3"
 env_logger = "0.11"

datafusion/physical-plan/Cargo.toml

@@ -78,6 +78,7 @@ datafusion-physical-expr-common = { workspace = true }
 datafusion-proto-common = { workspace = true, optional = true }
 datafusion-proto-models = { workspace = true, optional = true }
 futures = { workspace = true }
+genawaiter = { workspace = true, features = ["futures03"] }
 half = { workspace = true }
 hashbrown = { workspace = true }
 indexmap = { workspace = true }

datafusion/physical-plan/src/sorts/cursor.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use std::cmp::Ordering;
+use std::fmt::Debug;
 use std::sync::Arc;
 
 use arrow::array::{
@@ -32,7 +33,7 @@ use datafusion_execution::memory_pool::MemoryReservation;
 ///
 /// This is a trait as there are several specialized implementations, such as for
 /// single columns or for normalized multi column keys ([`Rows`])
-pub trait CursorValues {
+pub trait CursorValues: Debug + Sync + Send {
     fn len(&self) -> usize;
 
     /// Returns true if `l[l_idx] == r[r_idx]`
@@ -302,6 +303,7 @@ impl<T: ArrowNativeTypeOp> CursorValues for PrimitiveValues<T> {
     }
 }
 
+#[derive(Debug)]
 pub struct ByteArrayValues<T: OffsetSizeTrait> {
     offsets: OffsetBuffer<T>,
     values: Buffer,