Heap ও priority queue: top-k, merge ও task scheduling

Heap হলো priority queue-র পেছনের data structure: পুরো সংগ্রহ সাজিয়ে না রেখেও আপনি সবচেয়ে ছোট (বা বড়) item দ্রুত বের করতে পারেন। interview-তে ("top k", "k টি তালিকা মেলানো", "running median") এবং scheduler ও search-এর মতো আসল সিস্টেমে এটি আসে। নিচের কোড, test ও সময় প্রকাশের আগে Python 3-এ চালানো হয়েছে।

Min-heap কীভাবে কাজ করে

Min-heap হলো একটি binary tree যেখানে প্রতিটি parent তার সন্তানদের চেয়ে ছোট বা সমান। তাই সবচেয়ে ছোট item সবসময় মাথায়। এটি সাধারণ একটি list-এ রাখা হয়: i নম্বর জায়গার সন্তান 2i+1 ও 2i+2-এ, আর parent (i-1)//2-এ।

কাজ	খরচ	কীভাবে
ছোটতম দেখা	O(1)	০ নম্বর জায়গা পড়ুন
Push	O(log n)	শেষে যোগ করুন, parent-এর চেয়ে ছোট হলে উপরে সরান
ছোটতম বের করা (pop)	O(log n)	শেষ item মাথায় আনুন, পুরোনো মাথা সরান, নিচে নামান
n টি item থেকে বানানো	O(n)	নিচের অর্ধেক নিচ থেকে উপরে ঠিক করুন ("heapify")

Heap পুরোপুরি সাজানো নয়। শুধু parent-সন্তানের নিয়মটি মানে, আর ঠিক সেজন্যই পরের ছোটতম item দরকার হলে sort-এর চেয়ে এটি সস্তা।

একটি চালানো যায় এমন উদাহরণ

heaps.py নামে সংরক্ষণ করুন:

import heapq
import random
import time


class MinHeap:
    """A binary min-heap stored in a list: children of i are 2i+1 and 2i+2."""

    def __init__(self, items=()):
        self.a = list(items)
        for i in range(len(self.a) // 2 - 1, -1, -1):  # heapify in O(n)
            self._down(i)

    def __len__(self):
        return len(self.a)

    def peek(self):
        return self.a[0]

    def push(self, x):
        self.a.append(x)
        self._up(len(self.a) - 1)

    def pop(self):
        a = self.a
        a[0], a[-1] = a[-1], a[0]
        smallest = a.pop()
        if a:
            self._down(0)
        return smallest

    def _up(self, i):
        a = self.a
        while i > 0:
            parent = (i - 1) // 2
            if a[parent] <= a[i]:
                break
            a[parent], a[i] = a[i], a[parent]
            i = parent

    def _down(self, i):
        a, n = self.a, len(self.a)
        while True:
            left, right, small = 2 * i + 1, 2 * i + 2, i
            if left < n and a[left] < a[small]:
                small = left
            if right < n and a[right] < a[small]:
                small = right
            if small == i:
                return
            a[i], a[small] = a[small], a[i]
            i = small


def top_k_largest(items, k):
    """Keep a min-heap of size k: O(n log k) time, O(k) memory."""
    heap = []
    for x in items:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:
            heapq.heapreplace(heap, x)
    return sorted(heap, reverse=True)


def merge_sorted(lists):
    """Merge k sorted lists: the heap always holds one front item per list."""
    heap = [(lst[0], i, 0) for i, lst in enumerate(lists) if lst]
    heapq.heapify(heap)
    out = []
    while heap:
        value, i, j = heapq.heappop(heap)
        out.append(value)
        if j + 1 < len(lists[i]):
            heapq.heappush(heap, (lists[i][j + 1], i, j + 1))
    return out


class TaskQueue:
    """Lowest priority number first; a counter keeps equal priorities in insertion order."""

    def __init__(self):
        self._heap, self._n = [], 0

    def add(self, priority, task):
        heapq.heappush(self._heap, (priority, self._n, task))
        self._n += 1

    def next(self):
        return heapq.heappop(self._heap)[2]

    def __len__(self):
        return len(self._heap)


if __name__ == "__main__":
    h = MinHeap([9, 4, 7, 1, 8, 2])
    h.push(0)
    print("MinHeap pops :", [h.pop() for _ in range(len(h))])

    q = TaskQueue()
    for p, t in [(2, "send email"), (1, "fix outage"), (2, "write report"), (3, "tidy desk")]:
        q.add(p, t)
    print("task order   :", [q.next() for _ in range(len(q))])

    print("merge 3 lists:", merge_sorted([[1, 4, 9], [2, 3, 10], [5, 6, 7, 8]]))

    rnd = random.Random(1)
    data = [rnd.randint(0, 10**9) for _ in range(300_000)]
    t = time.perf_counter(); a = top_k_largest(data, 10); t1 = (time.perf_counter() - t) * 1000
    t = time.perf_counter(); b = sorted(data, reverse=True)[:10]; t2 = (time.perf_counter() - t) * 1000
    t = time.perf_counter(); c = heapq.nlargest(10, data); t3 = (time.perf_counter() - t) * 1000
    print(f"top 10 of 300,000: same answer = {a == b == c}")
    print(f"  heap of size k (ours)   {t1:7.1f} ms")
    print(f"  heapq.nlargest          {t3:7.1f} ms")
    print(f"  sorted(...)[:10]        {t2:7.1f} ms")

python3 heaps.py-এর ফল (সঠিক সময় মেশিনভেদে বদলায়):

MinHeap pops : [0, 1, 2, 4, 7, 8, 9]
task order   : ['fix outage', 'send email', 'write report', 'tidy desk']
merge 3 lists: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
top 10 of 300,000: same answer = True
  heap of size k (ours)      38.0 ms
  heapq.nlargest             12.7 ms
  sorted(...)[:10]          192.5 ms

কোড কী দেখায়

• MinHeap হাতে লেখা structure। সব item pop করলে সাজানো ক্রমে পাওয়া যায়, heap sort-এর ধারণাও এটাই।
• top_k_largest k আকারের একটি min-heap রাখে। এ পর্যন্ত রাখা k টির ছোটতমের চেয়ে বড় নতুন item এলে সেটিকে বদলে দেয়। সময় O(n log k), memory O(k), তাই memory-তে ধরে না এমন stream-এও চলে।
• merge_sorted প্রতিটি তালিকার সামনের একটি করে item heap-এ রেখে k টি সাজানো তালিকা মেলায়। প্রতি ধাপে O(log k), মোট n item মেলাতে O(n log k)।
• TaskQueue (priority, counter, task) রাখে। counter সমান priority-র মধ্যে ক্রম ঠিক রাখে (যে আগে এসেছে সে আগে বের হয়), আর Python-কে দুটি task সরাসরি তুলনা করতে হয় না।

সময় পড়া

৩,০০,০০০ সংখ্যা থেকে সবচেয়ে বড় ১০টি বের করতে:

• sorted(data, reverse=True)[:10] নিল ১৯২ ms, কারণ ১০টি রাখতে সব সাজাচ্ছে।
• আমাদের ১০ আকারের heap নিল ৩৮ ms, প্রায় ৫ গুণ দ্রুত।
• heapq.nlargest নিল ১২.৭ ms, আরও দ্রুত কারণ এটি C-তে লেখা। আমাদেরটি সাধারণ Python।

k যদি n-এর কাছাকাছি হয় তাহলে sort করাই যথেষ্ট। k ছোট আর n বড় হলে heap জেতে।

Python-এর heapq

Python-এ heapq আছে, একটি list-এর ওপর min-heap: heappush, heappop, heapify, heapreplace, nlargest, nsmallest। দুটি কথা মনে রাখুন:

• এটি শুধু min-heap। max-heap চাইলে সংখ্যাগুলোর ঋণাত্মক মান (বা ঋণাত্মক priority) রাখুন।
• item-গুলো তুলনাযোগ্য হতে হবে। tuple-এর ক্ষেত্রে Python প্রথম উপাদান তুলনা করে, তারপর পরেরটি, তাই ওপরের counter জরুরি।

Test

test_heaps.py নামে সংরক্ষণ করুন:

import heapq
import random
import unittest

from heaps import MinHeap, TaskQueue, merge_sorted, top_k_largest


class HeapTests(unittest.TestCase):
    def test_minheap_pops_in_order(self):
        rnd = random.Random(3)
        for _ in range(200):
            data = [rnd.randint(-20, 20) for _ in range(rnd.randint(0, 50))]
            h = MinHeap(data)
            self.assertEqual([h.pop() for _ in range(len(h))], sorted(data))

    def test_push_then_pop(self):
        h = MinHeap()
        for x in [5, 3, 8, 1]:
            h.push(x)
        self.assertEqual(h.peek(), 1)
        self.assertEqual([h.pop() for _ in range(4)], [1, 3, 5, 8])

    def test_top_k_matches_sorted(self):
        rnd = random.Random(5)
        for _ in range(100):
            data = [rnd.randint(0, 100) for _ in range(rnd.randint(0, 80))]
            for k in (1, 3, 10):
                self.assertEqual(top_k_largest(data, k), sorted(data, reverse=True)[:k])

    def test_merge_sorted(self):
        rnd = random.Random(9)
        lists = [sorted(rnd.randint(0, 50) for _ in range(rnd.randint(0, 10))) for _ in range(6)]
        self.assertEqual(merge_sorted(lists), sorted(x for l in lists for x in l))
        self.assertEqual(merge_sorted([[], []]), [])

    def test_task_queue_is_stable_for_equal_priority(self):
        q = TaskQueue()
        for t in ["a", "b", "c"]:
            q.add(1, t)
        self.assertEqual([q.next() for _ in range(3)], ["a", "b", "c"])


if __name__ == "__main__":
    unittest.main()

$ python3 -m unittest -v
Ran 5 tests in 0.025s

OK

প্রতিটি structure একই উত্তর পাওয়ার সহজ উপায়ের সঙ্গে মেলানো হয়েছে: heap, top-k ও merge-এর জন্য sorted()। random input-এ খালি তালিকা ও অনেক duplicate আছে।

কখন heap ব্যবহার করবেন

• বড় বা stream আসা সংগ্রহ থেকে top k বা k টি ছোটতম।
• k টি সাজানো উৎস মেলানো (log file, database shard, search ফল)।
• priority বা সবচেয়ে আগের সময় অনুযায়ী scheduling (timer, job queue, Dijkstra-র সংক্ষিপ্ত পথ)।
• Running median: নিচের অর্ধেকের জন্য একটি max-heap, ওপরের অর্ধেকের জন্য একটি min-heap।

মান ধরে item খুঁজতে হলে বা যেকোনো item দ্রুত সরাতে হলে heap ভুল যন্ত্র: খুঁজতে O(n) লাগে।

সাধারণ ভুল

• heap[1]-কে দ্বিতীয় ছোটতম ভাবা। শুধু heap[0]-এর নিশ্চয়তা আছে।
• সামান্য কয়েকটি top item পেতে সব sort করা।
• heapq-কে max-heap আশা করা।
• tie-breaker ছাড়া এমন tuple push করা যার দ্বিতীয় উপাদান তুলনা করা যায় না।
• heap ঠিক না করে ভেতরে কোনো item-এর priority বদলানো। নতুন entry push করুন, pop-এর সময় পুরোনোটি বাদ দিন।

মূল কথা: heap O(log n)-এ push ও pop দেয় আর ছোটতম item O(1)-এ। বারবার পরবর্তী সেরা item লাগলে, বিশেষ করে top-k ও merge-এ, এটি ব্যবহার করুন; আসল কোডে heapq ব্যবহার করুন।

---

Interview প্রস্তুতি ও DSA সিরিজ, পর্ব ৩ (আরও algorithm): 2 নম্বর (মোট 5)। আগের লেখা: Sorting algorithm: insertion, merge ও quick sort, সময়সহ.