Design Distributed Sorting Around an Explicit Helper Contract

Scenario

In a distributed storage system, data is spread across k worker nodes. You are tasked with implementing the coordinator for a globally ordered distributed sort (TeraSort-style). Instead of pulling all data to a single node, the coordinator must rely on an explicit helper contract exposed by each worker to sample, shuffle, and sort data in parallel.

Worker Helper Contract

Each worker implements the following interface:

class WorkerClient:
    def sample_keys(self, num_samples: int) -> list[tuple[int, str]]:
        """Returns a uniform random sample of (key, record_id) pairs from local storage."""
        pass

    def partition_and_send(self, boundaries: list[tuple[int, str]], worker_addresses: list[str]) -> None:
        """Partitions local records using `boundaries` (length k - 1) and transmits
        each partition to the corresponding target worker index [0..k-1]."""
        pass

    def local_sort_and_commit(self) -> int:
        """Sorts all received partitions in local memory/disk and finalizes.
        Returns the count of records held locally after sorting."""
        pass

Task

Implement the DistributedSortCoordinator class:

class DistributedSortCoordinator:
    def __init__(self, workers: list[WorkerClient]):
        self.workers = workers
        self.k = len(workers)

    def compute_boundaries(self, collected_samples: list[tuple[int, str]]) -> list[tuple[int, str]]:
        """Compute k - 1 splitter keys from all gathered samples to divide data into k balanced buckets."""
        pass

    def run_sort(self, samples_per_worker: int) -> list[int]:
        """Coordinates the full distributed sort workflow.
        Returns a list of record counts per worker [worker_0_count, ..., worker_k-1_count]."""
        pass

Requirements & Edge Cases

1. Total Global Ordering: Worker i must store only records strictly less than or equal to Worker i + 1's minimum record, determined by the composite tuple (key, record_id).
2. Duplicate Keys & Skew: Many records may share the exact same key. Using (key, record_id) ensures unique ordering and prevents partition overload when a single key has high frequency.
3. Balanced Partitions: With k partitions, boundary selection should pick splitters at approximately equal intervals across the sorted sample set.
4. Constraints: k >= 1. If k == 1, no boundaries are needed (boundaries = []). samples_per_worker >= 1.

Example

# 3 workers, target 2 boundaries (k - 1 = 2)
samples = [
    (10, "r1"), (10, "r2"), (10, "r3"),
    (20, "r4"), (30, "r5"), (40, "r6")
]
# Sorted samples: [(10, "r1"), (10, "r2"), (10, "r3"), (20, "r4"), (30, "r5"), (40, "r6")]
# Selecting splitters for 3 partitions (indices ~ 2, 4):
boundaries = [(10, "r3"), (30, "r5")]
# Partition 0 gets records  (10, "r3") and  (30, "r5")