feat: Implement a streaming merge sorted node

crates/polars-core/src/series/implementations/time.rs

@@ -156,7 +156,7 @@ impl SeriesTrait for SeriesWrap<TimeChunked> {
     }
     fn split_at(&self, offset: i64) -> (Series, Series) {
         let (a, b) = self.0.split_at(offset);
-        (a.into_series(), b.into_series())
+        (a.into_time().into_series(), b.into_time().into_series())
     }
 
     fn _sum_as_f64(&self) -> f64 {

crates/polars-stream/src/morsel.rs

@@ -16,6 +16,11 @@ pub fn get_ideal_morsel_size() -> usize {
     })
 }
 
+/// A token indicating the order of morsels in a stream.
+///
+/// The sequence tokens going through a pipe are monotonely non-decreasing and are allowed to be
+/// discontinuous. Consequently, `1 -> 1 -> 2` and `1 -> 3 -> 5` are valid streams of sequence
+/// tokens.
 #[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, Default)]
 pub struct MorselSeq(u64);