#74 merge_sorted_lists

Author: wojtask

src/solutions/chapter6/section5/exercise11.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import Generic
+
+from book.data_structures import Array
+from book.data_structures import KeyObject
+from book.data_structures import PriorityQueue
+from book.data_structures import T
+from solutions.chapter6.section5.exercise3 import min_heap_extract_min
+from solutions.chapter6.section5.exercise3 import min_heap_insert
+from util import range_of
+
+
+# TODO(#20) replace by the general list node and list classes once they will have been added
+class SortedListNode(Generic[T]):
+    key: T
+    next: SortedListNode[T] | None = None
+
+
+class SortedList(Generic[T]):
+    head: SortedListNode[T] | None = None
+
+
+def merge_sorted_lists(sorted_lists: Array[SortedList[T]], k: int) -> SortedList[T]:
+    """Merges sorted lists into one sorted list.
+
+    Args:
+        sorted_lists: an array of sorted lists
+        k: the number of sorted lists in the array
+
+    Returns:
+        The sorted list containing all elements from the input sorted lists.
+    """
+    Q = PriorityQueue(k)
+    for i in range_of(1, to=k):
+        sorted_list = sorted_lists[i]
+        if sorted_list.head is not None:
+            x = sorted_list.head
+            sorted_list.head = sorted_list.head.next
+            x.next = None
+            min_heap_insert(Q, KeyObject(key=x.key, data=(x, sorted_list)), k)
+    merged_list = SortedList[T]()
+    tail = None
+    while Q.heap_size > 0:
+        element, list_ = min_heap_extract_min(Q).data
+        if merged_list.head is None:
+            tail = merged_list.head = element
+        else:
+            tail.next = element
+            tail = tail.next
+        if list_.head is not None:
+            min_heap_insert(Q, KeyObject(key=list_.head.key, data=(list_.head, list_)), k)
+            list_.head = list_.head.next
+    return merged_list

test/test_solutions/test_chapter6.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import random
 from typing import Union
 
@@ -11,11 +13,15 @@
 from book.data_structures import CT
 from book.data_structures import Heap
 from book.data_structures import KeyObject
+from book.data_structures import T
 from solutions.chapter6.problem2 import multiary_child
 from solutions.chapter6.problem2 import multiary_parent
 from solutions.chapter6.section2.exercise3 import min_heapify
 from solutions.chapter6.section2.exercise6 import iterative_max_heapify
 from solutions.chapter6.section5.exercise10 import max_heap_delete
+from solutions.chapter6.section5.exercise11 import SortedList
+from solutions.chapter6.section5.exercise11 import SortedListNode
+from solutions.chapter6.section5.exercise11 import merge_sorted_lists
 from solutions.chapter6.section5.exercise3 import min_heap_decrease_key
 from solutions.chapter6.section5.exercise3 import min_heap_extract_min
 from solutions.chapter6.section5.exercise3 import min_heap_insert
@@ -28,6 +34,7 @@
 from solutions.chapter6.section5.exercise9 import min_heap_dequeue
 from solutions.chapter6.section5.exercise9 import min_heap_enqueue
 from test_case import ClrsTestCase
+from test_util import create_array
 from test_util import create_heap
 from test_util import create_priority_queue
 from util import range_of
@@ -44,6 +51,30 @@ def build_min_heap_by_inversion(A: Union[Heap[CT], Heap[KeyObject]], n: int) ->
             A[i] = -A[i]
 
 
+def create_sorted_list(elements: list[T]) -> SortedList[T]:
+    sortedList = SortedList[T]()
+    tail = None
+    for element in elements:
+        x = SortedListNode[T]()
+        x.key = element
+        if sortedList.head is None:
+            sortedList.head = x
+            tail = x
+        else:
+            tail.next = x
+            tail = x
+    return sortedList
+
+
+def convert_sorted_list(sorted_list: SortedList[T]) -> list[T]:
+    result = []
+    x = sorted_list.head
+    while x is not None:
+        result.append(x.key)
+        x = x.next
+    return result
+
+
 class TestChapter6(ClrsTestCase):
 
     @given(st.data())
@@ -261,6 +292,26 @@ def test_max_heap_increase_key_(self, data):
         self.assertArrayPermuted(A, key_objects, end=n)
         self.assertPriorityQueueMappingConsistent(A)
 
+    @given(st.data())
+    def test_max_heap_increase_key__invalid_key(self, data):
+        keys = data.draw(lists(integers(), min_size=1))
+        key_objects = [KeyObject(key, data.draw(text())) for key in keys]
+        heap = create_heap(key_objects)
+        n = len(key_objects)
+        build_max_heap(heap, n)
+        A = create_priority_queue(heap, n)
+        x = random.choice(key_objects)
+        k = x.key - data.draw(integers(min_value=1))
+
+        with self.assertRaisesRegex(ValueError, "new key is smaller than current key"):
+            max_heap_increase_key_(A, x, k)
+
+            self.assertNotEquals(x.key, k)
+            self.assertMaxHeap(A)
+            self.assertEqual(A.heap_size, n)
+            self.assertArrayPermuted(A, key_objects, end=n)
+            self.assertPriorityQueueMappingConsistent(A)
+
     @given(st.data())
     def test_min_heap_queue(self, data):
         # simulate the queue - enqueue, dequeue, then again enqueue and dequeue
@@ -350,24 +401,22 @@ def test_max_heap_delete(self, data):
         self.assertPriorityQueueMappingConsistent(A)
 
     @given(st.data())
-    def test_max_heap_increase_key__invalid_key(self, data):
-        keys = data.draw(lists(integers(), min_size=1))
-        key_objects = [KeyObject(key, data.draw(text())) for key in keys]
-        heap = create_heap(key_objects)
-        n = len(key_objects)
-        build_max_heap(heap, n)
-        A = create_priority_queue(heap, n)
-        x = random.choice(key_objects)
-        k = x.key - data.draw(integers(min_value=1))
-
-        with self.assertRaisesRegex(ValueError, "new key is smaller than current key"):
-            max_heap_increase_key_(A, x, k)
-
-            self.assertNotEquals(x.key, k)
-            self.assertMaxHeap(A)
-            self.assertEqual(A.heap_size, n)
-            self.assertArrayPermuted(A, key_objects, end=n)
-            self.assertPriorityQueueMappingConsistent(A)
+    def test_merge_sorted_lists(self, data):
+        k = data.draw(integers(min_value=1, max_value=10))
+        sorted_lists = [create_sorted_list(data.draw(lists(integers(), min_size=1, max_size=10).map(sorted)))
+                        for _ in range_of(1, to=k)]
+        sorted_lists_array = create_array(sorted_lists)
+        all_elements = []
+        for sorted_list in sorted_lists:
+            all_elements.extend(convert_sorted_list(sorted_list))
+
+        actual_merged_list = merge_sorted_lists(sorted_lists_array, k)
+
+        actual_all_elements = convert_sorted_list(actual_merged_list)
+        actual_all_elements_array = create_array(actual_all_elements)
+        n = len(all_elements)
+        self.assertArraySorted(actual_all_elements_array, end=n)
+        self.assertArrayPermuted(actual_all_elements_array, all_elements, end=n)
 
     @given(st.data())
     def test_multiary_parent_child(self, data):