refactor(rust): Fix slicing 0-width morsels on new streaming (#21940)

Author: nameexhaustion

crates/polars-stream/src/nodes/filter.rs

@@ -58,7 +58,7 @@ impl ComputeNode for FilterNode {
                         df._filter_seq(mask)
                     }).await?;
 
-                    if morsel.df().is_empty() {
+                    if morsel.df().height() == 0 {
                         continue;
                     }
 

crates/polars-stream/src/nodes/in_memory_source.rs

@@ -90,7 +90,7 @@ impl ComputeNode for InMemorySourceNode {
 
                     // TODO: remove this 'always sent at least one morsel'
                     // condition, see update_state.
-                    if df.is_empty() && seq > 0 {
+                    if df.height() == 0 && seq > 0 {
                         break;
                     }
 

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

@@ -150,7 +150,7 @@ impl SinkNode for PartedPartitionSinkNode {
                 let mut recv_port = recv_port.serial();
                 while let Ok(morsel) = recv_port.recv().await {
                     let (mut df, seq, source_token, consume_token) = morsel.into_inner();
-                    if df.is_empty() {
+                    if df.height() == 0 {
                         continue;
                     }
 

crates/polars-stream/src/nodes/io_sources/ndjson/negative_slice_pass.rs

@@ -195,7 +195,7 @@ impl MorselStreamReverser {
                             let mut df =
                                 combined_df.slice(row_offset.try_into().unwrap(), chunk_size);
 
-                            assert!(!df.is_empty()); // If we did our calculations properly
+                            assert!(df.height() > 0); // If we did our calculations properly
 
                             if let Some(row_index) = row_index.clone() {
                                 let offset = row_index.offset.saturating_add(

crates/polars-stream/src/nodes/merge_sorted.rs

@@ -56,7 +56,7 @@ fn find_mergeable(
 ) -> PolarsResult<Option<(DataFrame, DataFrame)>> {
     fn first_non_empty(vd: &mut VecDeque<DataFrame>) -> Option<DataFrame> {
         let mut df = vd.pop_front()?;
-        while df.is_empty() {
+        while df.height() == 0 {
             df = vd.pop_front()?;
         }
         Some(df)

crates/polars-stream/src/nodes/streaming_slice.rs

@@ -81,7 +81,7 @@ impl ComputeNode for StreamingSliceNode {
                     morsel.source_token().stop();
                 }
 
-                if !morsel.df().is_empty() && send.send(morsel).await.is_err() {
+                if morsel.df().height() > 0 && send.send(morsel).await.is_err() {
                     break;
                 }