fix(rust, python): take offset into account with str.explode

polars/polars-core/src/chunked_array/ops/explode.rs

@@ -463,13 +463,14 @@ impl ChunkExplode for Utf8Chunked {
             // capacity estimate
             let capacity = self.get_values_size() + validity.unset_bits();
 
+            let old_offsets = old_offsets.as_slice();
+            let mut old_offset = old_offsets[0];
             let mut new_offsets = Vec::with_capacity(capacity + 1);
-            new_offsets.push(0i64);
+            new_offsets.push(old_offset);
 
             let mut bitmap = MutableBitmap::with_capacity(capacity);
             let values = values.as_slice();
-            let mut old_offset = 0i64;
-            for (&offset, valid) in old_offsets.as_slice()[1..].iter().zip(validity) {
+            for (&offset, valid) in old_offsets[1..].iter().zip(validity) {
                 // safety:
                 // new_offsets already has a single value, so -1 is always in bounds
                 let latest_offset = unsafe { *new_offsets.get_unchecked(new_offsets.len() - 1) };
@@ -513,12 +514,13 @@ impl ChunkExplode for Utf8Chunked {
 
             // capacity estimate
             let capacity = self.get_values_size();
+            let old_offsets = old_offsets.as_slice();
+            let mut old_offset = old_offsets[0];
             let mut new_offsets = Vec::with_capacity(capacity + 1);
-            new_offsets.push(0i64);
+            new_offsets.push(old_offset);
 
             let values = values.as_slice();
-            let mut old_offset = 0i64;
-            for &offset in &old_offsets.as_slice()[1..] {
+            for &offset in &old_offsets[1..] {
                 // safety:
                 // new_offsets already has a single value, so -1 is always in bounds
                 let latest_offset = unsafe { *new_offsets.get_unchecked(new_offsets.len() - 1) };

polars/polars-lazy/src/physical_plan/planner/expr.rs

@@ -590,10 +590,9 @@ pub(crate) fn create_physical_expr(
         }
         Explode(expr) => {
             let input = create_physical_expr(expr, ctxt, expr_arena, schema)?;
-            let function = SpecialEq::new(Arc::new(move |s: &mut [Series]| {
-                let s = std::mem::take(&mut s[0]);
-                s.explode()
-            }) as Arc<dyn SeriesUdf>);
+            let function = SpecialEq::new(
+                Arc::new(move |s: &mut [Series]| s[0].explode()) as Arc<dyn SeriesUdf>
+            );
             Ok(Arc::new(ApplyExpr::new_minimal(
                 vec![input],
                 function,

py-polars/tests/unit/test_explode.py

@@ -224,3 +224,19 @@ def test_explode_inner_lists_3985() -> None:
         .agg(pl.col("categories"))
         .with_column(pl.col("categories").arr.eval(pl.element().arr.explode()))
     ).collect().to_dict(False) == {"id": [1], "categories": [["a", "b", "a", "c"]]}
+
+
+def test_utf8_sliced_explode() -> None:
+    df = pl.DataFrame(
+        {
+            "group": ["a", "b", "b"],
+            "values": ["foo", "bar", "baz"],
+        }
+    )
+
+    assert df.groupby("group", maintain_order=True).agg(
+        pl.col("values").flatten()
+    ).to_dict(False) == {
+        "group": ["a", "b"],
+        "values": [["f", "o", "o"], ["b", "a", "r", "b", "a", "z"]],
+    }