Fix performance regression in SourcePartitionedScheduler
`AbstractSet.removeAll(c)` performs terribly when `this.size() <= c.size()`
and `c.contains` is not `O(1)`.
This performance regression results in high coordinator CPU utilization
when a query scans a non-trivial number of splits (e.g. 1M splits in
30 minutes).
@@ -223,7 +223,7 @@ public class SourcePartitionedScheduler
splitAssignment = splitPlacementResult.getAssignments();
// remove splits with successful placements
- pendingSplits.removeAll(splitAssignment.values());
+ splitAssignment.values().forEach(pendingSplits::remove); // AbstractSet.removeAll performs terribly here.
overallSplitAssignmentCount += splitAssignment.size();
// if not completed placed, mark scheduleGroup as blocked on placement
