Correct stream ordered deallocation in Join (rapidsai#20981)

Our `Join.do_evaluate` function had a stream-ordering bug. We have 3 streams at play

1. left
2. right
3. result

The result depends on the left / right inputs, so we need to ensure that the deallocation of left / right happen after `result` is ready. We included a `join_cuda_streams` just before the `return` to ensure that it works. Unfortunately, we had reassigned `left` and `right` to new dataframes which were on the `result` stream, so all the streams were the same, so we didn't actually accomplish what we wanted.

The fix is to hold a reference to the original streams and use that.

This hopefully fixes a flaky test we've observed in cudf-polars with the rapidsmpf runtime
(https://github.com/rapidsai/cudf/actions/runs/20222481433/job/58046997204#step:12:1639).

Authors:
  - Tom Augspurger (https://github.com/TomAugspurger)

Approvers:
  - Bradley Dice (https://github.com/bdice)
  - Richard (Rick) Zamora (https://github.com/rjzamora)
  - Lawrence Mitchell (https://github.com/wence-)

URL: rapidsai#20981

Author: TomAugspurger

python/cudf_polars/cudf_polars/dsl/ir.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2024-2025, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
 """
 DSL nodes for the LogicalPlan of polars.
@@ -2384,8 +2384,14 @@ def do_evaluate(
         context: IRExecutionContext,
     ) -> DataFrame:
         """Evaluate and return a dataframe."""
+        # Save the original streams before any reassignments, since we need
+        # them for the final join_cuda_streams call to ensure proper stream
+        # ordering for deallocations.
+        original_left_stream = left.stream
+        original_right_stream = right.stream
         stream = get_joined_cuda_stream(
-            context.get_cuda_stream, upstreams=(left.stream, right.stream)
+            context.get_cuda_stream,
+            upstreams=(original_left_stream, original_right_stream),
         )
         how, nulls_equal, zlice, suffix, coalesce, maintain_order = options
         if how == "Cross":
@@ -2536,9 +2542,10 @@ def do_evaluate(
             result = result.slice(zlice)
 
         # Join the original streams back into the result stream to ensure that the
-        # deallocations (on the original streams) happen after the result is ready
+        # deallocations (on the original streams) happen after the result is ready.
         join_cuda_streams(
-            downstreams=(left.stream, right.stream), upstreams=(result.stream,)
+            downstreams=(original_left_stream, original_right_stream),
+            upstreams=(result.stream,),
         )
 
         return result