perf: Reduce sharing in stringview arrays in new-streaming equijoin (#…

…21129)

crates/polars-arrow/src/array/binview/mod.rs

@@ -449,6 +449,15 @@ impl<T: ViewType + ?Sized> BinaryViewArrayGeneric<T> {
         mutable.freeze().with_validity(self.validity)
     }
 
+    pub fn deshare(&self) -> Self {
+        if Arc::strong_count(&self.buffers) == 1
+            && self.buffers.iter().all(|b| b.storage_refcount() == 1)
+        {
+            return self.clone();
+        }
+        self.clone().gc()
+    }
+
     pub fn is_sliced(&self) -> bool {
         self.views.as_ptr() != self.views.storage_ptr()
     }

crates/polars-core/src/frame/mod.rs

@@ -601,6 +601,24 @@ impl DataFrame {
         }
     }
 
+    pub fn _deshare_views_mut(&mut self) {
+        // SAFETY: We never adjust the length or names of the columns.
+        unsafe {
+            let columns = self.get_columns_mut();
+            for col in columns {
+                let Column::Series(s) = col else { continue };
+
+                if let Ok(ca) = s.binary() {
+                    let gc_ca = ca.apply_kernel(&|a| a.deshare().into_boxed());
+                    *col = Column::from(gc_ca.into_series());
+                } else if let Ok(ca) = s.str() {
+                    let gc_ca = ca.apply_kernel(&|a| a.deshare().into_boxed());
+                    *col = Column::from(gc_ca.into_series());
+                }
+            }
+        }
+    }
+
     /// Rechunks all columns to only have a single chunk and turns it into a [`RecordBatchT`].
     pub fn rechunk_to_record_batch(
         self,

crates/polars-expr/src/expressions/window.rs

@@ -674,7 +674,7 @@ fn materialize_column(join_opt_ids: &ChunkJoinOptIds, out_column: &Column) -> Co
             Either::Left(ids) => unsafe {
                 IdxCa::with_nullable_idx(ids, |idx| out_column.take_unchecked(idx))
             },
-            Either::Right(ids) => unsafe { out_column.take_opt_chunked_unchecked(ids) },
+            Either::Right(ids) => unsafe { out_column.take_opt_chunked_unchecked(ids, false) },
         }
     }
 }