refactor(rust): Remove SyncCounter (#19556)

crates/polars-pipe/src/executors/sinks/slice.rs

@@ -1,43 +1,31 @@
 use std::any::Any;
-use std::sync::atomic::Ordering;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 
 use polars_core::error::PolarsResult;
 use polars_core::frame::DataFrame;
 use polars_core::schema::SchemaRef;
-use polars_utils::atomic::SyncCounter;
 
 use crate::operators::{
     chunks_to_df_unchecked, DataChunk, FinalizedSink, PExecutionContext, Sink, SinkResult,
 };
 
+#[derive(Clone)]
 // Ensure the data is return in the order it was streamed
 pub struct SliceSink {
-    offset: SyncCounter,
-    current_len: SyncCounter,
+    offset: Arc<AtomicUsize>,
+    current_len: Arc<AtomicUsize>,
     len: usize,
     chunks: Arc<Mutex<Vec<DataChunk>>>,
     schema: SchemaRef,
 }
 
-impl Clone for SliceSink {
-    fn clone(&self) -> Self {
-        Self {
-            offset: self.offset.clone(),
-            current_len: self.current_len.clone(),
-            len: self.len,
-            chunks: self.chunks.clone(),
-            schema: self.schema.clone(),
-        }
-    }
-}
-
 impl SliceSink {
     pub fn new(offset: u64, len: usize, schema: SchemaRef) -> SliceSink {
-        let offset = SyncCounter::new(offset as usize);
+        let offset = Arc::new(AtomicUsize::new(offset as usize));
         SliceSink {
             offset,
-            current_len: SyncCounter::new(0),
+            current_len: Arc::default(),
             len,
             chunks: Default::default(),
             schema,
@@ -101,18 +89,13 @@ impl Sink for SliceSink {
 
         let df = chunks_to_df_unchecked(chunks);
         let offset = self.offset.load(Ordering::Acquire) as i64;
-
-        // drop the counters
-        unsafe {
-            self.offset.manual_drop();
-            self.current_len.manual_drop();
-        }
-
         Ok(FinalizedSink::Finished(df.slice(offset, self.len)))
     }
+
     fn as_any(&mut self) -> &mut dyn Any {
         self
     }
+
     fn fmt(&self) -> &str {
         "slice_sink"
     }

crates/polars-utils/src/lib.rs

@@ -8,7 +8,6 @@
 pub mod abs_diff;
 pub mod algebraic_ops;
 pub mod arena;
-pub mod atomic;
 pub mod binary_search;
 pub mod cache;
 pub mod cardinality_sketch;