AlgoAtlas

Heap

Phase 2 12 solutions

Heap / Priority Queue

mindmap
  root((Heap))
    Top-K
      Kth Largest Element In a Stream LC703
      K Closest Points to Origin LC973
      Kth Largest Element In An Array LC215
    Two Heaps
      Find Median From Data Stream LC295
      IPO LC502
    Greedy + Heap
      Reorganize String LC767
      Longest Happy String LC1405
      Task Scheduler LC621
    Event Simulation
      Single Threaded CPU LC1834
      Car Pooling LC1094
    Design
      Design Twitter LC355

Problem List

# Problem LC Difficulty
1 Kth Largest Element In a Stream 703 🟢 Easy
2 Last Stone Weight 1046 🟢 Easy
3 K Closest Points to Origin 973 🟡 Medium
4 Kth Largest Element In An Array 215 🟡 Medium
5 Task Scheduler 621 🟡 Medium
6 Design Twitter 355 🟡 Medium
7 Single Threaded CPU 1834 🟡 Medium
8 Reorganize String 767 🟡 Medium
9 Longest Happy String 1405 🟡 Medium
10 Car Pooling 1094 🟡 Medium
11 Find Median From Data Stream 295 🔴 Hard
12 IPO 502 🔴 Hard

Approach Diagrams

Min Heap vs Max Heap Decision

flowchart TD
    A[Heap Problem] --> B{What do you need?}
    B -- Kth largest<br/>or top-K smallest --> C["Min Heap of size K<br/>heapq in Python"]
    B -- Kth smallest<br/>or top-K largest --> D["Max Heap of size K<br/>negate values in Python"]
    B -- Always need<br/>smallest next --> E["Min Heap<br/>LC 1834 CPU, 1094 Car Pooling"]
    B -- Always need<br/>largest next --> F["Max Heap<br/>LC 1046 Last Stone, 767 Reorganize"]
    B -- Need both halves --> G["Two Heaps<br/>LC 295 Median"]

Top-K Pattern Flow

flowchart TD
    A[Find Top-K elements] --> B[Use Min Heap of size K]
    B --> C[For each element x]
    C --> D{"heap size < K?"}
    D -- Yes --> E[Push x]
    D -- No --> F{"x > heap top?"}
    F -- Yes --> G[Pop min, push x]
    F -- No --> H[Skip]
    E --> C
    G --> C
    H --> C
    C --> I[Heap contains K largest]
    I --> J[LC 703, 215, 973]

Two Heaps Pattern — Median Finding

flowchart TD
    A[Add number n] --> B{"n <= max_heap top?"}
    B -- Yes --> C["Push to max_heap<br/>lower half"]
    B -- No --> D["Push to min_heap<br/>upper half"]
    C --> E{"Balance heaps<br/>size diff > 1?"}
    D --> E
    E -- max_heap bigger --> F["Move top of max_heap<br/>to min_heap"]
    E -- min_heap bigger --> G["Move top of min_heap<br/>to max_heap"]
    E -- balanced --> H[Done]
    F --> I[Get Median]
    G --> I
    H --> I
    I --> J{Equal sizes?}
    J -- Yes --> K[avg of both tops]
    J -- No --> L[top of larger heap]

Greedy + Heap Pattern

flowchart TD
    A[Greedy + Heap Problems] --> B{Pattern type}
    B -- Always pick most frequent --> C["Max Heap by frequency<br/>LC 767 Reorganize, 1405 Happy String"]
    C --> D["Pop max freq char<br/>append to result<br/>decrement and re-push"]
    B -- Unlock tasks by capital --> E["Two heaps: available vs locked<br/>LC 502 IPO"]
    E --> F["Min heap on cost<br/>Max heap on profit<br/>Unlock affordable, pick best profit"]
    B -- Cooldown scheduling --> G["Max heap + cooldown queue<br/>LC 621 Task Scheduler"]
    G --> H["Pop most frequent<br/>wait cooldown<br/>re-add to heap"]

Event-Driven Simulation with Heap

flowchart TD
    A[Event Simulation] --> B[Sort events by start time]
    B --> C["Min heap tracks active events<br/>by end time or priority"]
    C --> D[Process next event]
    D --> E{"Any events ended<br/>before current start?"}
    E -- Yes --> F["Pop from heap<br/>free resources"]
    E -- No --> G[All slots busy]
    F --> H[Assign to freed slot]
    G --> I["Wait / queue"]
    H --> J[Push new end time to heap]
    I --> J
    J --> D
    D --> K[LC 1834 CPU, 1094 Car Pooling]

Complexity Summary

Problem Time Space Pattern
Kth Largest in Stream (703) O(log k) per add O(k) Min heap size k
Last Stone Weight (1046) O(n log n) O(n) Max heap
K Closest Points (973) O(n log k) O(k) Min heap size k
Kth Largest in Array (215) O(n log k) O(k) Min heap size k
Task Scheduler (621) O(n log n) O(n) Max heap + cooldown
Design Twitter (355) O(k log n) per feed O(n) Min heap merge
Single Threaded CPU (1834) O(n log n) O(n) Min heap by duration
Reorganize String (767) O(n log k) O(k) Max heap by freq
Longest Happy String (1405) O(n log k) O(k) Max heap by freq
Car Pooling (1094) O(n log n) O(n) Min heap by dropoff
Find Median (295) O(log n) add, O(1) query O(n) Two heaps
IPO (502) O(n log n) O(n) Two heaps

Study Order

flowchart LR
    W1["Week 1: Heap Basics<br/>703 → 1046 → 215 → 973"]
    W2["Week 2: Greedy + Heap<br/>767 → 1405 → 621 → 502"]
    W3["Week 3: Simulation<br/>355 → 1834 → 1094"]
    W4["Week 4: Hard<br/>295"]
    W1 --> W2 --> W3 --> W4

Key insight: Python only has heapq (min heap). For max heap, negate values. For top-K largest, maintain a min heap of size K — if new element beats the min, swap it in.

Heap — Pattern Notes

Core Intuition

A heap gives you O(log n) insert and O(1) peek at the min (or max). Use it whenever you need to repeatedly extract the current best from a dynamic set. Python only has heapq (min-heap) — negate values for max-heap.

Sub-patterns:

  • Top-K — maintain a heap of size K; push and pop to keep only K best
  • Two Heaps — max-heap for lower half + min-heap for upper half → O(1) median
  • Greedy + Heap — sort by one dimension, use heap to greedily pick best on another
  • Event Simulation — push events by time/priority, process in order

Decision Diagram

flowchart TD
    A[Heap Problem] --> B{What do you need?}

    B --> C["K largest / smallest elements"]
    B --> D[Running median]
    B --> E["Schedule / assign tasks greedily"]
    B --> F[Simulate events in priority order]

    C --> C1["Min-heap of size K for K largest<br/>Max-heap of size K for K smallest"]
    D --> D1["Two heaps: max-heap left half<br/>min-heap right half<br/>balance sizes on each insert"]
    E --> E1["Sort input, use heap to track<br/>best available option greedily"]
    F --> F1["Push all events; pop by priority<br/>handle dependencies by pushing new events"]

Per-Problem Notes

# Problem LC Difficulty What to Learn Common Mistakes Time / Space
1 Kth Largest in Stream 703 Easy Min-heap of size K; heap[0] is always the Kth largest Heap size > K — must pop when size exceeds K O(log k) add / O(k) space
2 Last Stone Weight 1046 Easy Max-heap (negate); pop two heaviest, push difference if nonzero Forgetting to push back when stones aren't equal O(n log n) / O(n)
3 K Closest Points 973 Medium Min-heap by distance; pop K times. Or max-heap of size K Computing distance with sqrt (unnecessary — compare squared) O(n log k) / O(k)
4 Kth Largest in Array 215 Medium Min-heap of size K; final heap[0] is answer. Or quickselect O(n) avg Using max-heap and popping n-k times (works but slower) O(n log k) / O(k)
5 Task Scheduler 621 Medium Greedy: always schedule most frequent available task. Use max-heap + cooldown queue Not handling idle time correctly; forgetting cooldown queue O(n log n) / O(n)
6 Design Twitter 355 Medium Per-user tweet list; on getNewsFeed merge K lists with max-heap of (timestamp, user, idx) Not limiting to 10 tweets; heap comparison on equal timestamps O(f log f) getNewsFeed / O(n)
7 Single-Threaded CPU 1834 Medium Sort tasks by enqueue time; use min-heap (processing_time, idx); simulate time jumps Not jumping time forward when CPU is idle O(n log n) / O(n)
8 Reorganize String 767 Medium Max-heap by frequency; greedily pick most frequent, then second most frequent Placing same char consecutively; not re-adding char with remaining count O(n log k) / O(k)
9 Longest Happy String 1405 Medium Same greedy as Reorganize String but allow up to 2 consecutive; pick 2nd most freq when top would make 3 Not switching to second most frequent when top would violate constraint O(n log k) / O(k)
10 Car Pooling 1094 Medium Event simulation: push (time, +/-passengers); sort/heap by time; track current passengers Not handling simultaneous drop-off before pick-up at same stop O(n log n) / O(n)
11 Find Median from Stream 295 Hard Two heaps: max-heap (lo) + min-heap (hi); balance so len(lo) == len(hi) or len(lo) == len(hi)+1 Not rebalancing after each insert; median formula for even/odd sizes O(log n) add, O(1) median / O(n)
12 IPO 502 Hard Sort projects by capital; greedily unlock affordable projects into max-heap by profit; pick top K times Not re-checking newly affordable projects after each pick O(n log n) / O(n)

Edge Cases to Watch

  • K larger than array size — return all elements
  • All stones equal weight in Last Stone — result is 0 (even count) or that weight (odd count)
  • Task Scheduler with all unique tasks — no idle time needed
  • Find Median with all same elements — median is that element
  • Car Pooling: drop-off and pick-up at same location — drop off first
  • IPO with 0 initial capital — only projects with capital=0 are initially available
  • Reorganize String impossible — when max_freq > (n+1)//2, return ""
  • Two heaps: always push to lo first, then rebalance to hi if needed
# Problem LC Difficulty
1 Kth Largest Element In a Stream 703 🟢 Easy
2 Last Stone Weight 1046 🟢 Easy
3 K Closest Points to Origin 973 🟡 Medium
4 Kth Largest Element In An Array 215 🟡 Medium
5 Task Scheduler 621 🟡 Medium
6 Design Twitter 355 🟡 Medium
7 Single Threaded CPU 1834 🟡 Medium
8 Reorganize String 767 🟡 Medium
9 Longest Happy String 1405 🟡 Medium
10 Car Pooling 1094 🟡 Medium
11 Find Median From Data Stream 295 🔴 Hard
12 IPO 502 🔴 Hard
Car Pooling â–¼ expand 
import heapq
from typing import List


class Solution:
    """
    ## 1094. Car Pooling

    There is a car with `capacity` empty seats. The vehicle only drives east (i.e., it cannot turn
    around and drive west).

    You are given the integer `capacity` and an array `trips` where `trips[i] = [numPassengersi,
    fromi, toi]` indicates that the `i`th trip has `numPassengersi` passengers and the route goes
    from `fromi` to `toi`. The locations are given as the number of kilometers due east from the
    car's initial location.

    Return `true` if it is possible to pick up and drop off all passengers for all the given trips,
    or `false` otherwise.

    ### Examples

    ```text
    Input: trips = [[2,1,5],[3,3,7]], capacity = 4
    Output: false

    Explanation:
        At stop 1: pick up 2 → current = 2
        At stop 3: pick up 3 → current = 5 > 4 → impossible
    ```

    ```text
    Input: trips = [[2,1,5],[3,3,7]], capacity = 5
    Output: true
    ```

    ```text
    Input: trips = [[3,2,7],[3,7,9],[8,3,9]], capacity = 11
    Output: true
    ```

    ### Constraints
    - `1 <= trips.length <= 1000`
    - `trips[i].length == 3`
    - `1 <= numPassengersi <= 100`
    - `0 <= fromi < toi <= 1000`
    - `1 <= capacity <= 10^5`
    """

    def carPooling_events(self, trips: List[List[int]], capacity: int) -> bool:
        # O(n log n) — sort pickup/dropoff events, simulate
        events = []
        for num, start, end in trips:
            events.append((start, num))   # pickup
            events.append((end, -num))    # dropoff
        events.sort()
        current = 0
        for _, delta in events:
            current += delta
            if current > capacity:
                return False
        return True

    def carPooling(self, trips: List[List[int]], capacity: int) -> bool:
        # Sort trips by start; use min-heap of (end, passengers) for active trips
        # At each trip start, drop off passengers from trips that have ended
        # If current passengers exceed capacity, return False
        # O(n + max_stop) — difference array over stop positions
        diff = [0] * 1001
        for num, start, end in trips:
            diff[start] += num
            diff[end] -= num
        current = 0
        for delta in diff:
            current += delta
            if current > capacity:
                return False
        return True


if __name__ == '__main__':
    s = Solution()
    assert s.carPooling_events([[2, 1, 5], [3, 3, 7]], 4) == False
    assert s.carPooling([[2, 1, 5], [3, 3, 7]], 4) == False

    assert s.carPooling_events([[2, 1, 5], [3, 3, 7]], 5) == True
    assert s.carPooling([[2, 1, 5], [3, 3, 7]], 5) == True

    assert s.carPooling([[3, 2, 7], [3, 7, 9], [8, 3, 9]], 11) == True
    assert s.carPooling_events([[3, 2, 7], [3, 7, 9], [8, 3, 9]], 11) == True

    print("All tests passed.")
Design Twitter â–¼ expand 
import heapq
from collections import defaultdict
from typing import List


class Twitter:
    """
    ## 355. Design Twitter

    Design a simplified version of Twitter where users can post tweets, follow/unfollow another user,
    and see the 10 most recent tweets in the user's news feed.

    Implement the `Twitter` class:
    - `Twitter()` — initializes the Twitter object.
    - `void postTweet(int userId, int tweetId)` — composes a new tweet with ID `tweetId` by user `userId`.
    - `List<Integer> getNewsFeed(int userId)` — retrieves the 10 most recent tweet IDs in the user's
      news feed. Each item must be posted by users who the user followed or by the user themselves.
      Tweets must be ordered from most recent to least recent.
    - `void follow(int followerId, int followeeId)` — the user with ID `followerId` starts following
      the user with ID `followeeId`.
    - `void unfollow(int followerId, int followeeId)` — the user with ID `followerId` starts unfollowing
      the user with ID `followeeId`.

    ### Examples

    ```text
    Input:
        Twitter twitter = new Twitter();
        twitter.postTweet(1, 5);
        twitter.getNewsFeed(1);   // [5]
        twitter.follow(1, 2);
        twitter.postTweet(2, 6);
        twitter.getNewsFeed(1);   // [6, 5]
        twitter.unfollow(1, 2);
        twitter.getNewsFeed(1);   // [5]
    ```

    ### Constraints
    - `1 <= userId, tweetId <= 500`
    - All the tweets have **unique** IDs.
    - At most `3 * 10^4` calls will be made to `postTweet`, `getNewsFeed`, `follow`, and `unfollow`.
    """

    def __init__(self):
        # O(1) init
        self.timestamp = 0
        self.tweets: dict = defaultdict(list)   # userId -> [(time, tweetId), ...]
        self.following: dict = defaultdict(set)  # userId -> {followeeId}

    def postTweet(self, userId: int, tweetId: int) -> None:
        # Store tweet with a global timestamp for ordering
        # O(1) — append tweet with global timestamp
        self.tweets[userId].append((self.timestamp, tweetId))
        self.timestamp += 1

    def getNewsFeed(self, userId: int) -> List[int]:
        # Collect recent tweets from user and all followees
        # Use heap or sort to get the 10 most recent
        # O(f * 10 * log f) — heap-merge latest tweets from followed users + self
        heap: List = []
        users = self.following[userId] | {userId}
        for uid in users:
            if self.tweets[uid]:
                idx = len(self.tweets[uid]) - 1
                time, tid = self.tweets[uid][idx]
                heapq.heappush(heap, (-time, tid, uid, idx))

        result = []
        while heap and len(result) < 10:
            neg_time, tid, uid, idx = heapq.heappop(heap)
            result.append(tid)
            if idx > 0:
                time, next_tid = self.tweets[uid][idx - 1]
                heapq.heappush(heap, (-time, next_tid, uid, idx - 1))
        return result

    def follow(self, followerId: int, followeeId: int) -> None:
        # Add followee to user's follow set
        # O(1)
        self.following[followerId].add(followeeId)

    def unfollow(self, followerId: int, followeeId: int) -> None:
        # Remove followee from user's follow set if present
        # O(1)
        self.following[followerId].discard(followeeId)


if __name__ == '__main__':
    t = Twitter()
    t.postTweet(1, 5)
    assert t.getNewsFeed(1) == [5]
    t.follow(1, 2)
    t.postTweet(2, 6)
    assert t.getNewsFeed(1) == [6, 5]
    t.unfollow(1, 2)
    assert t.getNewsFeed(1) == [5]

    t2 = Twitter()
    for i in range(11):
        t2.postTweet(1, i)
    feed = t2.getNewsFeed(1)
    assert len(feed) == 10
    assert feed == list(range(10, 0, -1))

    print("All tests passed.")
Find Median From Data Stream â–¼ expand 
import heapq
from typing import List


class MedianFinder:
    """
    ## 295. Find Median from Data Stream

    The **median** is the middle value in an ordered integer list. If the size of the list is even,
    there is no middle value, and the median is the mean of the two middle values.

    Implement the `MedianFinder` class:
    - `MedianFinder()` — initializes the `MedianFinder` object.
    - `void addNum(int num)` — adds the integer `num` from the data stream to the data structure.
    - `double findMedian()` — returns the median of all elements so far. Answers within `10^-5`
      of the actual answer will be accepted.

    ### Examples

    ```text
    Input:
        MedianFinder mf = new MedianFinder();
        mf.addNum(1);
        mf.addNum(2);
        mf.findMedian(); // 1.5
        mf.addNum(3);
        mf.findMedian(); // 2.0
    ```

    ### Constraints
    - `-10^5 <= num <= 10^5`
    - There will be at least one element in the data structure before calling `findMedian`.
    - At most `5 * 10^4` calls will be made to `addNum` and `findMedian`.
    """

    # --- Approach 1: Sort on every findMedian ---

    class _SortApproach:
        def __init__(self):
            self.data: List[int] = []

        def addNum(self, num: int) -> None:
            # Push to max-heap (lower half); rebalance if max of lower > min of upper
            # Keep sizes balanced: small can have at most 1 more element than large
            # O(1) amortized append
            import bisect
            bisect.insort(self.data, num)

        def findMedian(self) -> float:
            # Odd total: median is top of larger heap; even: average of both tops
            # O(1) — data kept sorted via insort
            n = len(self.data)
            return self.data[n // 2] if n % 2 else (self.data[n // 2 - 1] + self.data[n // 2]) / 2.0

    # --- Approach 2: Two heaps ---

    def __init__(self):
        # O(1) init
        self.small: List[int] = []  # max-heap (negated) — lower half
        self.large: List[int] = []  # min-heap — upper half

    def addNum(self, num: int) -> None:
        # Push to max-heap (lower half); rebalance if max of lower > min of upper
        # Keep sizes balanced: small can have at most 1 more element than large
        # O(log n) — push to max-heap, rebalance
        heapq.heappush(self.small, -num)
        # Ensure every element in small <= every element in large
        if self.small and self.large and (-self.small[0]) > self.large[0]:
            heapq.heappush(self.large, -heapq.heappop(self.small))
        # Balance sizes: small can have at most 1 more than large
        if len(self.small) > len(self.large) + 1:
            heapq.heappush(self.large, -heapq.heappop(self.small))
        elif len(self.large) > len(self.small):
            heapq.heappush(self.small, -heapq.heappop(self.large))

    def findMedian(self) -> float:
        # Odd total: median is top of larger heap; even: average of both tops
        # O(1) — peek tops of both heaps
        if len(self.small) > len(self.large):
            return float(-self.small[0])
        return (-self.small[0] + self.large[0]) / 2.0


if __name__ == '__main__':
    mf = MedianFinder()
    mf.addNum(1)
    mf.addNum(2)
    assert mf.findMedian() == 1.5
    mf.addNum(3)
    assert mf.findMedian() == 2.0

    mf2 = MedianFinder()
    mf2.addNum(-1)
    assert mf2.findMedian() == -1.0
    mf2.addNum(-2)
    assert mf2.findMedian() == -1.5
    mf2.addNum(-3)
    assert mf2.findMedian() == -2.0
    mf2.addNum(-4)
    assert mf2.findMedian() == -2.5
    mf2.addNum(-5)
    assert mf2.findMedian() == -3.0

    print("All tests passed.")
Ipo â–¼ expand 
import heapq
from typing import List


class Solution:
    """
    ## 502. IPO

    Suppose LeetCode will start its **IPO** soon. In order to sell a good price of its shares to
    Venture Capital, LeetCode would like to work on some projects to increase its capital before
    the IPO. Since it has limited resources, it can only finish at most `k` distinct projects
    before the IPO. Help LeetCode design the best way to maximize its total capital after finishing
    at most `k` distinct projects.

    You are given `n` projects where the `i`th project has a pure profit `profits[i]` and a minimum
    capital of `capital[i]` is needed to start it.

    Initially, you have `w` capital. When you finish a project, you will obtain its pure profit and
    the profit will be added to your total capital.

    Return the **maximized total capital** after finishing at most `k` distinct projects.

    ### Examples

    ```text
    Input: k = 2, w = 0, profits = [1, 2, 3], capital = [0, 1, 1]
    Output: 4

    Explanation:
        Start with w=0. Available: project 0 (capital=0). Pick it → profit=1, w=1.
        Now w=1. Available: projects 1,2 (capital=1). Pick project 2 (profit=3) → w=4.
        Result: 4
    ```

    ```text
    Input: k = 3, w = 0, profits = [1, 2, 3], capital = [0, 1, 2]
    Output: 6
    ```

    ### Constraints
    - `1 <= k <= 10^5`
    - `0 <= w <= 10^9`
    - `n == profits.length == capital.length`
    - `1 <= n <= 10^5`
    - `0 <= profits[i] <= 10^4`
    - `0 <= capital[i] <= 10^9`
    """

    def findMaximizedCapital(self, k: int, w: int, profits: List[int], capital: List[int]) -> int:
        # Greedily pick the most profitable available project at each step
        # Use min-heap on capital to unlock projects as capital grows
        # Use max-heap on profit to always pick the best available project
        # O(n log n) — sort by capital, greedily pick max profit available project
        projects = sorted(zip(capital, profits))
        available: List[int] = []  # max-heap (negated profits)
        i = 0
        n = len(projects)

        for _ in range(k):
            # Unlock all projects we can now afford
            while i < n and projects[i][0] <= w:
                heapq.heappush(available, -projects[i][1])
                i += 1
            if not available:
                break
            w += -heapq.heappop(available)

        return w


if __name__ == '__main__':
    s = Solution()
    assert s.findMaximizedCapital(2, 0, [1, 2, 3], [0, 1, 1]) == 4
    assert s.findMaximizedCapital(3, 0, [1, 2, 3], [0, 1, 2]) == 6
    assert s.findMaximizedCapital(1, 0, [1, 2, 3], [1, 1, 2]) == 0  # can't afford any
    assert s.findMaximizedCapital(2, 1, [1, 2, 3], [1, 1, 2]) == 6
    print("All tests passed.")
K Closest Points To Origin â–¼ expand 
import heapq
import random
from typing import List


class Solution:
    """
    ## 973. K Closest Points to Origin

    Given an array of `points` where `points[i] = [xi, yi]` represents a point on the X-Y plane
    and an integer `k`, return the `k` closest points to the origin `(0, 0)`.

    The distance between two points on the X-Y plane is the Euclidean distance:
    `sqrt((x1 - x2)^2 + (y1 - y2)^2)`.

    You may return the answer in **any order**. The answer is **guaranteed** to be unique
    (except for the order that it is in).

    ### Examples

    ```text
    Input: points = [[1, 3], [-2, 2]], k = 1
    Output: [[-2, 2]]

    Explanation:
        Distance of [1,3] from origin = sqrt(10) ≈ 3.16
        Distance of [-2,2] from origin = sqrt(8) ≈ 2.83
        [-2,2] is closer, so return [[-2,2]].
    ```

    ```text
    Input: points = [[3, 3], [5, -1], [-2, 4]], k = 2
    Output: [[3, 3], [-2, 4]]
    ```

    ### Constraints
    - `1 <= k <= points.length <= 10^4`
    - `-10^4 <= xi, yi <= 10^4`
    """

    def kClosest_sort(self, points: List[List[int]], k: int) -> List[List[int]]:
        # O(n log n) — sort all points by squared distance
        points.sort(key=lambda p: p[0] ** 2 + p[1] ** 2)
        return points[:k]

    def kClosest_heap(self, points: List[List[int]], k: int) -> List[List[int]]:
        # O(n log k) — max-heap of size k using negated distances
        heap: List = []
        for x, y in points:
            dist = -(x * x + y * y)
            heapq.heappush(heap, (dist, x, y))
            if len(heap) > k:
                heapq.heappop(heap)
        return [[x, y] for _, x, y in heap]

    def kClosest(self, points: List[List[int]], k: int) -> List[List[int]]:
        # Use max-heap of size k; evict farthest point when heap exceeds k
        # No need to sort — heap maintains k closest points
        # O(n) average — quickselect partitioning by squared distance
        def dist(p):
            return p[0] ** 2 + p[1] ** 2

        def partition(lo, hi):
            pivot = dist(points[hi])
            i = lo
            for j in range(lo, hi):
                if dist(points[j]) <= pivot:
                    points[i], points[j] = points[j], points[i]
                    i += 1
            points[i], points[hi] = points[hi], points[i]
            return i

        lo, hi = 0, len(points) - 1
        while lo < hi:
            mid = partition(lo, hi)
            if mid == k - 1:
                break
            elif mid < k - 1:
                lo = mid + 1
            else:
                hi = mid - 1
        return points[:k]


if __name__ == '__main__':
    s = Solution()

    res = s.kClosest_sort([[1, 3], [-2, 2]], 1)
    assert res == [[-2, 2]]

    res = s.kClosest_heap([[1, 3], [-2, 2]], 1)
    assert sorted(res) == [[-2, 2]]

    res = s.kClosest([[1, 3], [-2, 2]], 1)
    assert res == [[-2, 2]]

    res = s.kClosest([[3, 3], [5, -1], [-2, 4]], 2)
    assert sorted(res) == [[-2, 4], [3, 3]]

    print("All tests passed.")
Kth Largest Element In A Stream â–¼ expand 
import heapq
from typing import List


class KthLargest:
    """
    ## 703. Kth Largest Element in a Stream

    Design a class to find the `k`th largest element in a stream.
    Note that it is the `k`th largest element in the sorted order, not the `k`th distinct element.

    Implement the `KthLargest` class:
    - `KthLargest(int k, int[] nums)` — initializes the object with `k` and the stream `nums`.
    - `int add(int val)` — appends `val` to the stream and returns the element representing
      the `k`th largest element in the stream.

    ### Examples

    ```text
    Input:
        KthLargest(3, [4, 5, 8, 2])
        add(3) -> 4
        add(5) -> 5
        add(10) -> 8
        add(9) -> 8
        add(4) -> 8

    Explanation:
        After init: stream = [2, 4, 5, 8], k=3 → 3rd largest = 4
        add(3): stream = [2, 3, 4, 5, 8] → 3rd largest = 4
        add(5): stream = [2, 3, 4, 5, 5, 8] → 3rd largest = 5
    ```

    ### Constraints
    - `1 <= k <= 10^4`
    - `0 <= nums.length <= 10^4`
    - `-10^4 <= nums[i] <= 10^4`
    - `-10^4 <= val <= 10^4`
    - At most `10^4` calls to `add`
    - It is guaranteed that there will be at least `k` elements in the array when `add` is called.
    """

    # --- Approach 1: Sort on every add ---

    def __init__sort(self, k: int, nums: List[int]):
        # O(n log n) init
        self.k = k
        self.nums = sorted(nums)

    def add_sort(self, val: int) -> int:
        # O(n log n) per add — insert and re-sort
        import bisect
        bisect.insort(self.nums, val)
        return self.nums[-self.k]

    # --- Approach 2: Min-heap of size k ---

    def __init__(self, k: int, nums: List[int]):
        # Maintain a min-heap of size k; top is always the k-th largest
        # O(n log k) init
        self.k = k
        self.heap: List[int] = []
        for n in nums:
            heapq.heappush(self.heap, n)
            if len(self.heap) > k:
                heapq.heappop(self.heap)

    def add(self, val: int) -> int:
        # Push new val; if heap exceeds k, pop the smallest (not in top-k)
        # Top of min-heap is the k-th largest element
        # O(log k) per add — maintain min-heap of size k
        heapq.heappush(self.heap, val)
        if len(self.heap) > self.k:
            heapq.heappop(self.heap)
        return self.heap[0]


if __name__ == '__main__':
    # stream after init: [2,4,5,8], k=3 → 3rd largest = 4
    obj = KthLargest(3, [4, 5, 8, 2])
    assert obj.add(3) == 4   # [2,3,4,5,8] → 3rd largest = 4
    assert obj.add(5) == 5   # [2,3,4,5,5,8] → 3rd largest = 5
    assert obj.add(10) == 5  # [2,3,4,5,5,8,10] → 3rd largest = 5
    assert obj.add(9) == 8   # [2,3,4,5,5,8,9,10] → 3rd largest = 8
    assert obj.add(4) == 8   # [2,3,4,4,5,5,8,9,10] → 3rd largest = 8

    obj2 = KthLargest(1, [])
    assert obj2.add(-3) == -3
    assert obj2.add(-2) == -2
    assert obj2.add(-4) == -2

    print("All tests passed.")
Kth Largest Element In An Array â–¼ expand 
import heapq
from typing import List


class Solution:
    """
    ## 215. Kth Largest Element in an Array

    Given an integer array `nums` and an integer `k`, return the `k`th largest element in the array.

    Note that it is the `k`th largest element in the sorted order, not the `k`th distinct element.

    You must solve it without sorting (for the optimal solution).

    ### Examples

    ```text
    Input: nums = [3, 2, 1, 5, 6, 4], k = 2
    Output: 5
    ```

    ```text
    Input: nums = [3, 2, 3, 1, 2, 4, 5, 5, 6], k = 4
    Output: 4
    ```

    ### Constraints
    - `1 <= k <= nums.length <= 10^5`
    - `-10^4 <= nums[i] <= 10^4`
    """

    def findKthLargest_sort(self, nums: List[int], k: int) -> int:
        # O(n log n) — sort descending and index
        return sorted(nums, reverse=True)[k - 1]

    def findKthLargest_heap(self, nums: List[int], k: int) -> int:
        # O(n log k) — min-heap of size k
        heap: List[int] = []
        for n in nums:
            heapq.heappush(heap, n)
            if len(heap) > k:
                heapq.heappop(heap)
        return heap[0]

    def findKthLargest(self, nums: List[int], k: int) -> int:
        # Min-heap of size k: after processing all elements, top is k-th largest
        # O(n) average — quickselect; target index in sorted-asc = len-k
        target = len(nums) - k

        def partition(lo, hi):
            pivot, fill = nums[hi], lo
            for i in range(lo, hi):
                if nums[i] <= pivot:
                    nums[fill], nums[i] = nums[i], nums[fill]
                    fill += 1
            nums[fill], nums[hi] = nums[hi], nums[fill]
            return fill

        lo, hi = 0, len(nums) - 1
        while lo < hi:
            mid = partition(lo, hi)
            if mid == target:
                break
            elif mid < target:
                lo = mid + 1
            else:
                hi = mid - 1
        return nums[target]


if __name__ == '__main__':
    s = Solution()
    assert s.findKthLargest_sort([3, 2, 1, 5, 6, 4], 2) == 5
    assert s.findKthLargest_heap([3, 2, 1, 5, 6, 4], 2) == 5
    assert s.findKthLargest([3, 2, 1, 5, 6, 4], 2) == 5

    assert s.findKthLargest_sort([3, 2, 3, 1, 2, 4, 5, 5, 6], 4) == 4
    assert s.findKthLargest_heap([3, 2, 3, 1, 2, 4, 5, 5, 6], 4) == 4
    assert s.findKthLargest([3, 2, 3, 1, 2, 4, 5, 5, 6], 4) == 4

    print("All tests passed.")
Last Stone Weight â–¼ expand 
import heapq
from typing import List


class Solution:
    """
    ## 1046. Last Stone Weight

    You are given an array of integers `stones` where `stones[i]` is the weight of the `i`th stone.

    We are playing a game with the stones. On each turn, we choose the **two heaviest** stones and
    smash them together. Suppose the heaviest two stones have weights `x` and `y` with `x <= y`:
    - If `x == y`, both stones are destroyed.
    - If `x != y`, the stone of weight `x` is destroyed, and the stone of weight `y` has new weight `y - x`.

    At the end of the game, there is **at most one** stone left. Return the weight of the last remaining
    stone. If there are no stones left, return `0`.

    ### Examples

    ```text
    Input: stones = [2, 7, 4, 1, 8, 1]
    Output: 1

    Explanation:
        Smash 8 and 7 → 1. Stones: [2, 4, 1, 1, 1]
        Smash 4 and 2 → 2. Stones: [2, 1, 1, 1]
        Smash 2 and 1 → 1. Stones: [1, 1, 1]
        Smash 1 and 1 → 0. Stones: [1]
        Result: 1
    ```

    ```text
    Input: stones = [1]
    Output: 1
    ```

    ### Constraints
    - `1 <= stones.length <= 30`
    - `1 <= stones[i] <= 1000`
    """

    def lastStoneWeight_sort(self, stones: List[int]) -> int:
        # O(n^2 log n) — sort on every iteration
        stones = sorted(stones)
        while len(stones) > 1:
            y, x = stones.pop(), stones.pop()
            if x != y:
                import bisect
                bisect.insort(stones, y - x)
        return stones[0] if stones else 0

    def lastStoneWeight(self, stones: List[int]) -> int:
        # Use max-heap (negate for Python's min-heap)
        # Repeatedly smash two heaviest; push remainder if non-zero
        # O(n log n) — max-heap via negation
        heap = [-s for s in stones]
        heapq.heapify(heap)
        while len(heap) > 1:
            y = -heapq.heappop(heap)
            x = -heapq.heappop(heap)
            if x != y:
                heapq.heappush(heap, -(y - x))
        return -heap[0] if heap else 0


if __name__ == '__main__':
    s = Solution()
    assert s.lastStoneWeight([2, 7, 4, 1, 8, 1]) == 1
    assert s.lastStoneWeight([1]) == 1
    assert s.lastStoneWeight([2, 2]) == 0
    assert s.lastStoneWeight_sort([2, 7, 4, 1, 8, 1]) == 1
    assert s.lastStoneWeight_sort([1]) == 1
    print("All tests passed.")
Longest Happy String â–¼ expand 
import heapq


class Solution:
    """
    ## 1405. Longest Happy String

    A string `s` is called **happy** if it satisfies all of the following conditions:
    - `s` only contains the letters `'a'`, `'b'`, and `'c'`.
    - `s` does not contain `"aaa"`, `"bbb"`, or `"ccc"` as a substring.
    - `s` contains **at most** `a` occurrences of the letter `'a'`.
    - `s` contains **at most** `b` occurrences of the letter `'b'`.
    - `s` contains **at most** `c` occurrences of the letter `'c'`.

    Given three integers `a`, `b`, and `c`, return the **longest possible happy string**.
    If there are multiple longest happy strings, return any of them. If there is no such string,
    return the empty string `""`.

    ### Examples

    ```text
    Input: a = 1, b = 1, c = 7
    Output: "ccbccacc"
    ```

    ```text
    Input: a = 2, b = 2, c = 1
    Output: "aabbc" (or any valid)
    ```

    ```text
    Input: a = 7, b = 1, c = 0
    Output: "aabaa"
    ```

    ### Constraints
    - `0 <= a, b, c <= 100`
    - `a + b + c > 0`
    """

    def longestDiverseString(self, a: int, b: int, c: int) -> str:
        # Greedily pick the most frequent character that doesn't create 3 consecutive
        # If top char would create 3 in a row, use second most frequent instead
        # Re-add characters to heap after use if count > 0
        # O(a+b+c) — greedy: always pick most frequent char, limited to 2 consecutive
        heap = []
        for cnt, ch in [(-a, 'a'), (-b, 'b'), (-c, 'c')]:
            if cnt < 0:
                heapq.heappush(heap, (cnt, ch))

        result = []
        while heap:
            cnt, ch = heapq.heappop(heap)
            # If last two chars are same as current, use second most frequent
            if len(result) >= 2 and result[-1] == result[-2] == ch:
                if not heap:
                    break
                cnt2, ch2 = heapq.heappop(heap)
                result.append(ch2)
                if cnt2 + 1 < 0:
                    heapq.heappush(heap, (cnt2 + 1, ch2))
                heapq.heappush(heap, (cnt, ch))
            else:
                result.append(ch)
                if cnt + 1 < 0:
                    heapq.heappush(heap, (cnt + 1, ch))

        return ''.join(result)


if __name__ == '__main__':
    s = Solution()

    def is_happy(res, a, b, c):
        if 'aaa' in res or 'bbb' in res or 'ccc' in res:
            return False
        if res.count('a') > a or res.count('b') > b or res.count('c') > c:
            return False
        return True

    res = s.longestDiverseString(1, 1, 7)
    assert is_happy(res, 1, 1, 7) and len(res) == 8, f"Got '{res}'"

    res = s.longestDiverseString(7, 1, 0)
    assert is_happy(res, 7, 1, 0) and len(res) == 5, f"Got '{res}'"

    res = s.longestDiverseString(2, 2, 1)
    assert is_happy(res, 2, 2, 1) and len(res) == 5, f"Got '{res}'"

    print("All tests passed.")
Reorganize String â–¼ expand 
import heapq
from collections import Counter


class Solution:
    """
    ## 767. Reorganize String

    Given a string `s`, rearrange the characters of `s` so that any two adjacent characters are
    not the same. Return any possible rearrangement of `s` or return `""` if not possible.

    ### Examples

    ```text
    Input: s = "aab"
    Output: "aba"
    ```

    ```text
    Input: s = "aaab"
    Output: ""
    ```

    ```text
    Input: s = "vvvlo"
    Output: "vlvov" (or any valid rearrangement)
    ```

    ### Constraints
    - `1 <= s.length <= 500`
    - `s` consists of lowercase English letters.
    """

    def reorganizeString(self, s: str) -> str:
        # Always place the most frequent remaining character
        # Use max-heap; after placing, cool down for one step before re-adding
        # O(n log k) where k=26 — greedy: always place most frequent char not equal to last placed
        counts = Counter(s)
        heap = [(-cnt, ch) for ch, cnt in counts.items()]
        heapq.heapify(heap)

        result = []
        prev_cnt, prev_ch = 0, ''

        while heap:
            cnt, ch = heapq.heappop(heap)
            result.append(ch)
            # Re-push previous character now that it's no longer adjacent
            if prev_cnt < 0:
                heapq.heappush(heap, (prev_cnt, prev_ch))
            prev_cnt, prev_ch = cnt + 1, ch  # cnt is negative, so +1 decrements

        return ''.join(result) if len(result) == len(s) else ''


if __name__ == '__main__':
    s = Solution()

    res = s.reorganizeString("aab")
    assert res == "aba", f"Got {res}"

    res = s.reorganizeString("aaab")
    assert res == ""

    res = s.reorganizeString("a")
    assert res == "a"

    # Validate no two adjacent chars are same for longer inputs
    res = s.reorganizeString("vvvlo")
    assert res and all(res[i] != res[i + 1] for i in range(len(res) - 1))

    res = s.reorganizeString("aabbcc")
    assert res and all(res[i] != res[i + 1] for i in range(len(res) - 1))

    print("All tests passed.")
Single Threaded Cpu â–¼ expand 
import heapq
from typing import List


class Solution:
    """
    ## 1834. Single-Threaded CPU

    You are given `n` tasks labeled from `0` to `n - 1` represented by a 2D integer array `tasks`,
    where `tasks[i] = [enqueueTimei, processingTimei]` represents the time at which the `i`th task
    will be available for processing and the time it takes to finish processing.

    Return the **order** in which the CPU will process the tasks.

    Rules:
    - If the CPU is idle and there are no available tasks, the CPU waits until the next available task.
    - If the CPU is idle and there are available tasks, the CPU picks the task with the **shortest
      processing time**. If multiple tasks have the same processing time, it picks the task with the
      **smallest index**.
    - Once a task is started, the CPU runs it to completion.

    ### Examples

    ```text
    Input: tasks = [[1,2],[2,4],[3,2],[4,1]]
    Output: [0, 2, 3, 1]

    Explanation:
        t=1: task 0 available, CPU picks task 0 (proc=2), finishes at t=3
        t=3: tasks 1,2 available; pick task 2 (proc=2, shorter than 4), finishes at t=5
        t=5: tasks 1,3 available; pick task 3 (proc=1), finishes at t=6
        t=6: task 1 available; pick task 1, finishes at t=10
    ```

    ```text
    Input: tasks = [[7,10],[7,12],[7,5],[7,4],[7,2]]
    Output: [4, 3, 2, 0, 1]
    ```

    ### Constraints
    - `1 <= tasks.length <= 10^5`
    - `1 <= enqueueTimei, processingTimei <= 10^9`
    """

    def getOrder(self, tasks: List[List[int]]) -> List[int]:
        # Sort tasks by enqueue time; simulate CPU with a min-heap of (processing_time, index)
        # At each step, advance time to next available task if CPU is idle
        # Always process the shortest available task (min-heap by processing time, then index)
        # O(n log n) — sort by enqueue time, simulate with min-heap on (proc_time, index)
        indexed = sorted(enumerate(tasks), key=lambda x: x[1][0])
        heap: List = []  # (processing_time, original_index)
        result = []
        time = 0
        i = 0
        n = len(tasks)

        while len(result) < n:
            # Push all tasks available at current time
            while i < n and indexed[i][1][0] <= time:
                orig_idx, task = indexed[i]
                heapq.heappush(heap, (task[1], orig_idx))
                i += 1

            if heap:
                proc_time, orig_idx = heapq.heappop(heap)
                result.append(orig_idx)
                time += proc_time
            else:
                # CPU idle — jump to next task's enqueue time
                time = indexed[i][1][0]

        return result


if __name__ == '__main__':
    s = Solution()
    assert s.getOrder([[1, 2], [2, 4], [3, 2], [4, 1]]) == [0, 2, 3, 1]
    assert s.getOrder([[7, 10], [7, 12], [7, 5], [7, 4], [7, 2]]) == [4, 3, 2, 0, 1]
    assert s.getOrder([[1, 1], [1, 1]]) == [0, 1]
    print("All tests passed.")
Task Scheduler â–¼ expand 
import heapq
from collections import Counter, deque
from typing import List


class Solution:
    """
    ## 621. Task Scheduler

    Given a characters array `tasks`, representing the tasks a CPU needs to do, where each letter
    represents a different task. Tasks could be done in any order. Each task is done in one unit of
    time. For each unit of time, the CPU could complete either one task or just be idle.

    However, there is a non-negative integer `n` that represents the cooldown interval between two
    **same tasks** (the same letter in the array), that is that there must be at least `n` units of
    time between any two same tasks.

    Return the **least number of units of times** that the CPU will take to finish all the given tasks.

    ### Examples

    ```text
    Input: tasks = ["A","A","A","B","B","B"], n = 2
    Output: 8

    Explanation: A -> B -> idle -> A -> B -> idle -> A -> B
    ```

    ```text
    Input: tasks = ["A","A","A","B","B","B"], n = 0
    Output: 6
    ```

    ```text
    Input: tasks = ["A","A","A","A","A","A","B","C","D","E","F","G"], n = 2
    Output: 16
    ```

    ### Constraints
    - `1 <= tasks.length <= 10^4`
    - `tasks[i]` is an uppercase English letter.
    - `0 <= n <= 100`
    """

    def leastInterval_heap(self, tasks: List[str], n: int) -> int:
        # O(t * n) — simulate with max-heap and cooldown queue
        counts = Counter(tasks)
        heap = [-c for c in counts.values()]
        heapq.heapify(heap)
        time = 0
        cooldown: deque = deque()  # (available_time, neg_count)
        while heap or cooldown:
            time += 1
            if heap:
                cnt = heapq.heappop(heap) + 1  # decrement count
                if cnt < 0:
                    cooldown.append((time + n, cnt))
            if cooldown and cooldown[0][0] == time:
                heapq.heappush(heap, cooldown.popleft()[1])
        return time

    def leastInterval(self, tasks: List[str], n: int) -> int:
        # Most frequent task determines idle time needed
        # Formula: (max_freq - 1) * (n + 1) + count_of_max_freq_tasks
        # But if tasks fill all slots, answer is just len(tasks)
        # O(t) — math formula based on most frequent task
        counts = Counter(tasks)
        max_count = max(counts.values())
        num_max = sum(1 for c in counts.values() if c == max_count)
        return max(len(tasks), (max_count - 1) * (n + 1) + num_max)


if __name__ == '__main__':
    s = Solution()
    assert s.leastInterval_heap(["A", "A", "A", "B", "B", "B"], 2) == 8
    assert s.leastInterval(["A", "A", "A", "B", "B", "B"], 2) == 8

    assert s.leastInterval_heap(["A", "A", "A", "B", "B", "B"], 0) == 6
    assert s.leastInterval(["A", "A", "A", "B", "B", "B"], 0) == 6

    assert s.leastInterval(["A", "A", "A", "A", "A", "A", "B", "C", "D", "E", "F", "G"], 2) == 16

    print("All tests passed.")

Heap Visualizer

Insert (sift-up) and Extract-Min (sift-down)

3851210
Array:3851210
Min-Heap. Insert a value or Extract Min to see sift operations.