add test and cleanup synccounters

polars/polars-lazy/polars-pipe/src/executors/sinks/sort/sink.rs

@@ -154,6 +154,12 @@ impl Sink for SortSink {
     }
 
     fn finalize(&mut self, _context: &PExecutionContext) -> PolarsResult<FinalizedSink> {
+        // safety: we are the final thread and will drop only once.
+        unsafe {
+            self.mem_total.manual_drop();
+            self.free_mem.manual_drop();
+        }
+
         if self.ooc {
             let lock = self.io_thread.lock().unwrap();
             let io_thread = lock.as_ref().unwrap();

polars/polars-lazy/src/physical_plan/streaming/convert.rs

@@ -352,7 +352,6 @@ pub(crate) fn insert_streaming_nodes(
                 continue;
             }
             let verbose = std::env::var("POLARS_VERBOSE").is_ok();
-            dbg!(verbose);
 
             for branch in tree {
                 // should be reset for every branch

py-polars/tests/slow/test_streaming.py

@@ -1,3 +1,4 @@
+import os
 import time
 
 import numpy as np
@@ -13,3 +14,21 @@ def test_cross_join_stack() -> None:
     assert a.join(a, how="cross").head().collect(streaming=True).shape == (5, 2)
     t1 = time.time()
     assert (t1 - t0) < 0.5
+
+
+def test_ooc_sort() -> None:
+    # not sure if monkeypatch will be visible in rust
+    env = "POLARS_FORCE_OOC_SORT"
+    os.environ[env] = "1"
+
+    s = pl.arange(0, 100_000, eager=True).rename("idx")
+
+    df = s.shuffle().to_frame()
+
+    for reverse in [True, False]:
+        out = (
+            df.lazy().sort("idx", reverse=reverse).collect(streaming=True)
+        ).to_series()
+
+        assert out.series_equal(s.sort(reverse=reverse))
+    os.unsetenv(env)