perf: Drop tasks early in distributed limits (#6350)

Author: srilman

src/daft-distributed/src/pipeline_node/limit.rs

@@ -328,7 +328,10 @@ impl LimitNode {
 
             // Update max_concurrent_tasks based on actual output
             // Only update if we have remaining limit, and we did get some output
-            if !limit_state.is_take_done() && total_num_rows > 0 && num_local_limits > 0 {
+            if limit_state.is_take_done() {
+                // Drop the input channel to cancel any input tasks
+                break;
+            } else if total_num_rows > 0 && num_local_limits > 0 {
                 let rows_per_task = total_num_rows.div_ceil(num_local_limits);
                 max_concurrent_tasks = limit_state.remaining_take().div_ceil(rows_per_task);
             }