Sorting in Practice — Library Functions

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# Your "simple" quicksort implementation:
def my_quicksort(arr):
    if len(arr) <= 1:
        return arr
    pivot = arr[len(arr) // 2]
    left = [x for x in arr if x < pivot]
    middle = [x for x in arr if x == pivot]
    right = [x for x in arr if x > pivot]
    return my_quicksort(left) + middle + my_quicksort(right)
 
# Things your implementation probably doesn't handle:
# ❌ Worst-case O(n²) for sorted input
# ❌ Worst-case O(n²) for all-equal input (depending on implementation)
# ❌ O(n) extra space (not in-place)
# ❌ Stack overflow for deeply nested recursion
# ❌ Not stable
# ❌ Doesn't switch to insertion sort for small subarrays
# ❌ No median-of-three pivot selection
# ❌ No introspection to heapsort fallback
# ❌ Cache-oblivious optimizations
# ❌ Handling of NaN, infinity, and special float values
# ❌ Memory allocation failures
# ❌ Comparison function exceptions
 
# What Python's built-in does:
# ✅ Timsort: O(n) best case for partially sorted data
# ✅ O(n log n) guaranteed worst case
# ✅ Stable sorting
# ✅ Adaptive to input patterns
# ✅ Optimized C implementation
# ✅ Galloping mode for efficient merges
# ✅ Handles all edge cases correctly
 
# The one-line equivalent:
sorted_arr = sorted(arr)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

import heapq
import random
import time
 
# Generate 10 million random ratings
n = 10_000_000
products = [(random.random() * 5, f"product_{i}") for i in range(n)]
 
# APPROACH 1: Full sort (what many developers would do)
start = time.time()
sorted_products = sorted(products, key=lambda x: x[0], reverse=True)[:20]
full_sort_time = time.time() - start
print(f"Full sort: {full_sort_time:.3f}s")
 
# APPROACH 2: Partial sort with heapq (the right way for top-k)
start = time.time()
top_20 = heapq.nlargest(20, products, key=lambda x: x[0])
partial_time = time.time() - start
print(f"heapq.nlargest: {partial_time:.3f}s")
 
print(f"Speedup: {full_sort_time/partial_time:.1f}x")
 
# Typical results:
# Full sort: 4.5s
# heapq.nlargest: 1.2s
# Speedup: 3.7x
 
# For even better performance when k is very small, 
# you can use a selection algorithm (quickselect):
 
def quickselect_k_largest(arr, k, key=lambda x: x):
    """
    Find the k largest elements using quickselect.
    Expected O(n) time complexity.
    """
    if len(arr) <= k:
        return sorted(arr, key=key, reverse=True)
    
    # For production use, consider using numpy.partition or similar
    arr = arr.copy()
    
    def partition(left, right, pivot_idx):
        pivot_val = key(arr[pivot_idx])
        arr[right], arr[pivot_idx] = arr[pivot_idx], arr[right]
        store_idx = left
        for i in range(left, right):
            if key(arr[i]) > pivot_val:  # Greater for largest
                arr[store_idx], arr[i] = arr[i], arr[store_idx]
                store_idx += 1
        arr[store_idx], arr[right] = arr[right], arr[store_idx]
        return store_idx
    
    def select(left, right, k):
        while True:
            if left == right:
                return
            pivot_idx = random.randint(left, right)
            pivot_idx = partition(left, right, pivot_idx)
            if k == pivot_idx:
                return
            elif k < pivot_idx:
                right = pivot_idx - 1
            else:
                left = pivot_idx + 1
    
    select(0, len(arr) - 1, k - 1)
    return sorted(arr[:k], key=key, reverse=True)
 
 
# Libraries that provide optimized partial sorting:
# - C++: std::partial_sort, std::nth_element
# - Python: heapq.nsmallest, heapq.nlargest
# - Go: sort.Slice with early exit (manual)
# - Rust: slice.select_nth_unstable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

"""
External Merge Sort Implementation
 
Use this when sorting data that doesn't fit in RAM.
For production, consider using database ORDER BY, pandas external sort,
or purpose-built tools like Unix sort -m.
"""
 
import os
import heapq
import tempfile
from typing import Iterator, List, Callable, Any
 
def external_sort(
    input_file: str,
    output_file: str,
    key: Callable = lambda x: x,
    chunk_size_mb: int = 100,
    temp_dir: str = None
) -> None:
    """
    Sort a file that doesn't fit in memory.
    
    Args:
        input_file: Path to input file (one item per line)
        output_file: Path to write sorted output
        key: Key function for comparison
        chunk_size_mb: Approximate memory to use per chunk
        temp_dir: Directory for temporary files
    """
    # Estimate lines per chunk based on memory budget
    # This is approximate; real implementations would be smarter
    sample_lines = []
    with open(input_file, 'r') as f:
        for i, line in enumerate(f):
            sample_lines.append(line)
            if i >= 100:
                break
    
    avg_line_size = sum(len(line) for line in sample_lines) / len(sample_lines)
    lines_per_chunk = int((chunk_size_mb * 1024 * 1024) / avg_line_size)
    
    sorted_chunks: List[str] = []
    temp_dir = temp_dir or tempfile.gettempdir()
    
    # PHASE 1: Create sorted chunks
    print("Phase 1: Creating sorted chunks...")
    chunk_count = 0
    
    with open(input_file, 'r') as f:
        while True:
            # Read a chunk
            chunk = []
            for line in f:
                chunk.append(line.rstrip('\n'))
                if len(chunk) >= lines_per_chunk:
                    break
            
            if not chunk:
                break
            
            # Sort the chunk in memory (using library sort!)
            chunk.sort(key=key)
            
            # Write to temporary file
            chunk_file = os.path.join(temp_dir, f'sort_chunk_{chunk_count}.tmp')
            with open(chunk_file, 'w') as cf:
                for line in chunk:
                    cf.write(line + '\n')
            
            sorted_chunks.append(chunk_file)
            chunk_count += 1
            print(f"  Created chunk {chunk_count} ({len(chunk)} lines)")
    
    # PHASE 2: K-way merge
    print(f"Phase 2: Merging {len(sorted_chunks)} chunks...")
    
    def chunk_iterator(filepath: str) -> Iterator[str]:
        """Iterate over lines in a chunk file."""
        with open(filepath, 'r') as f:
            for line in f:
                yield line.rstrip('\n')
    
    # Open all chunks and create a heap of (value, chunk_index, iterator)
    chunk_iters = [chunk_iterator(path) for path in sorted_chunks]
    
    # Initialize heap with first element from each chunk
    heap = []
    for i, it in enumerate(chunk_iters):
        try:
            value = next(it)
            heapq.heappush(heap, (key(value), i, value, it))
        except StopIteration:
            pass  # Empty chunk
    
    # Merge and write output
    with open(output_file, 'w') as out:
        while heap:
            _, chunk_idx, value, it = heapq.heappop(heap)
            out.write(value + '\n')
            
            # Get next element from the same chunk
            try:
                next_value = next(it)
                heapq.heappush(heap, (key(next_value), chunk_idx, next_value, it))
            except StopIteration:
                pass  # This chunk is exhausted
    
    # Cleanup temporary files
    print("Cleaning up temporary files...")
    for chunk_file in sorted_chunks:
        os.remove(chunk_file)
    
    print(f"Done! Sorted output written to {output_file}")
 
 
# Example usage:
# external_sort(
#     'huge_log_file.txt',
#     'sorted_log_file.txt',
#     key=lambda line: line.split()[0],  # Sort by first field
#     chunk_size_mb=500
# )
 
# For production use, prefer:
# - Database: SELECT ... ORDER BY ... (databases have optimized external sort)
# - pandas: df.to_csv() with sort_values (uses external temp files)
# - Unix: sort -m sorted_chunk1 sorted_chunk2 ... > output
# - Apache Spark: df.orderBy() (distributed sorting)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

import random
import time
 
def counting_sort(arr, max_val):
    """
    Sort integers in range [0, max_val].
    Time: O(n + max_val)
    Space: O(max_val)
    """
    counts = [0] * (max_val + 1)
    
    # Count occurrences
    for val in arr:
        counts[val] += 1
    
    # Reconstruct sorted array
    result = []
    for val, count in enumerate(counts):
        result.extend([val] * count)
    
    return result
 
 
def radix_sort_lsd(arr, max_digits=None):
    """
    Radix sort using LSD (Least Significant Digit) approach.
    Time: O(d × n) where d is number of digits
    Space: O(n)
    """
    if not arr:
        return arr
    
    if max_digits is None:
        max_digits = len(str(max(arr)))
    
    for digit_pos in range(max_digits):
        # Use counting sort as stable subroutine for each digit
        buckets = [[] for _ in range(10)]
        divisor = 10 ** digit_pos
        
        for num in arr:
            digit = (num // divisor) % 10
            buckets[digit].append(num)
        
        arr = []
        for bucket in buckets:
            arr.extend(bucket)
    
    return arr
 
 
# Benchmark: Sorting ages (0-100 range)
n = 10_000_000
ages = [random.randint(0, 100) for _ in range(n)]
ages_copy = ages.copy()
 
# Standard library sort
start = time.time()
sorted_ages_lib = sorted(ages)
lib_time = time.time() - start
 
# Counting sort
start = time.time()
sorted_ages_counting = counting_sort(ages_copy, 100)
counting_time = time.time() - start
 
print(f"Library sort:  {lib_time:.2f}s")
print(f"Counting sort: {counting_time:.2f}s")
print(f"Speedup:       {lib_time/counting_time:.1f}x")
 
# Typical results for 10M integers in [0, 100]:
# Library sort:  3.5s
# Counting sort: 0.7s
# Speedup:       5.0x
 
# For larger ranges, the advantage diminishes.
# Crossover point: when max_val approaches n × log(n),
# comparison sort becomes competitive.
 
 
# Benchmark: Sorting 7-digit product codes
product_codes = [random.randint(1000000, 9999999) for _ in range(1_000_000)]
codes_copy = product_codes.copy()
 
start = time.time()
sorted_codes_lib = sorted(product_codes)
lib_time = time.time() - start
 
start = time.time()
sorted_codes_radix = radix_sort_lsd(codes_copy, max_digits=7)
radix_time = time.time() - start
 
print(f"\nProduct codes (7 digits):")
print(f"Library sort: {lib_time:.2f}s")
print(f"Radix sort:   {radix_time:.2f}s")
print(f"Speedup:      {lib_time/radix_time:.1f}x")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

START: Do you need to sort data?
  │
  ├─ YES
  │    │
  │    ▼
  │    Does the standard library sort work for your use case?
  │    (array.sort(), Arrays.sort(), std::sort, etc.)
  │      │
  │      ├─ YES → USE LIBRARY SORT ✓
  │      │         You're done. Don't overthink it.
  │      │
  │      └─ NO → Why doesn't it work?
  │               │
  │               ├─ Need partial sort (top-k)?
  │               │    │
  │               │    └─ Does library have partial_sort / nlargest?
  │               │         ├─ YES → USE LIBRARY PARTIAL SORT ✓
  │               │         └─ NO → Implement heap-based selection
  │               │
  │               ├─ Data doesn't fit in memory?
  │               │    │
  │               │    ├─ Can you use database ORDER BY? → USE DATABASE ✓
  │               │    ├─ Can you use Spark/distributed tool? → USE THAT ✓
  │               │    └─ None available → Implement external merge sort
  │               │
  │               ├─ Data has special structure (small range, fixed digits)?
  │               │    │
  │               │    ├─ Is performance actually a problem?
  │               │    │    ├─ NO → USE LIBRARY SORT ✓
  │               │    │    └─ YES → Benchmark counting/radix sort
  │               │    │              Is it actually faster for your data?
  │               │    │                ├─ YES → Use non-comparison sort
  │               │    │                └─ NO → USE LIBRARY SORT ✓
  │               │
  │               ├─ Need stability and library is unstable?
  │               │    │
  │               │    ├─ Is there a stable_sort variant? → USE THAT ✓
  │               │    └─ NO → Use index trick with library sort
  │               │
  │               ├─ Specialized hardware (GPU, SIMD)?
  │               │    │
  │               │    └─ Is this a genuine bottleneck justified by profiling?
  │               │         ├─ NO → USE LIBRARY SORT ✓
  │               │         └─ YES → Use hardware-specific library (CUB, etc.)
  │               │                   or implement carefully
  │               │
  │               └─ Other exotic requirement?
  │                    │
  │                    └─ Have you truly exhausted library options?
  │                         ├─ NO → Research more; there's usually something
  │                         └─ YES → Implement with extensive testing
  │
  └─ NO: You don't need to sort
         │
         └─ Wait, what are you actually trying to do?
              │
              ├─ Find min/max? → Use argmin/argmax, O(n)
              ├─ Find median? → Use quickselect, O(n)
              ├─ Find top-k? → Use heap, O(n log k)
              ├─ Check if sorted? → Single pass, O(n)
              └─ Something else → Sorting might not be the answer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

# The complaint: "sorted() is too slow for my data"
 
# The "slow" code:
def process_data_slow(items):
    # Sort by item score (this seems slow!)
    sorted_items = sorted(items, key=lambda x: x.calculate_score())
    return sorted_items[:10]
 
# Analysis: Where is the time actually going?
import time
 
# If calculate_score() is expensive (e.g., database query, complex math)
# it gets called O(n log n) times during sorting!
 
# Example: expensive score calculation
class Item:
    def __init__(self, data):
        self.data = data
        self._cached_score = None
    
    def calculate_score(self):
        # Simulating expensive operation (0.001s per call)
        time.sleep(0.001)  
        return sum(self.data) / len(self.data)
 
items = [Item([i, i+1, i+2]) for i in range(100)]
 
# This calls calculate_score() O(n log n) ≈ 700 times!
# 700 × 0.001s = 0.7 seconds just in score calculations
 
start = time.time()
sorted_items = sorted(items, key=lambda x: x.calculate_score())
print(f"Slow sort: {time.time() - start:.2f}s")
 
 
# THE FIX: Cache the expensive computation, not replace the sort algorithm!
 
def process_data_fast(items):
    # Pre-compute scores once: O(n) calls
    scored_items = [(item.calculate_score(), item) for item in items]
    # Sort by pre-computed scores
    scored_items.sort(key=lambda x: x[0])
    # Extract items
    return [item for score, item in scored_items[:10]]
 
# Or use Python's key function caching:
from functools import lru_cache
 
class ItemWithCaching:
    def __init__(self, data):
        self.data = data
        self._data_tuple = tuple(data)  # Make hashable for caching
    
    @lru_cache(maxsize=None)
    def calculate_score(self):
        return sum(self.data) / len(self.data)
 
# Now sorted() calls calculate_score() O(n) times, not O(n log n)
# Because Python caches key values before sorting!
 
# The lesson: The "sorting bottleneck" was actually a 
# "comparison function bottleneck". The fix was caching, 
# not implementing a custom sorting algorithm.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

"""
If you implement a custom sort, test it like this.
This example uses hypothesis for property-based testing.
"""
 
from hypothesis import given, strategies as st, settings
import random
 
# Your custom sort implementation
def my_custom_sort(arr, key=lambda x: x):
    """Custom radix sort for integers in known range."""
    # ... implementation ...
    pass  # Placeholder
 
 
# === PROPERTY-BASED TESTS ===
 
@given(st.lists(st.integers(min_value=-1000, max_value=1000)))
def test_result_is_sorted(arr):
    """Output must be in sorted order."""
    result = my_custom_sort(arr.copy())
    for i in range(len(result) - 1):
        assert result[i] <= result[i + 1], f"Not sorted at index {i}"
 
 
@given(st.lists(st.integers()))
def test_length_preserved(arr):
    """Output must have same length as input."""
    result = my_custom_sort(arr.copy())
    assert len(result) == len(arr)
 
 
@given(st.lists(st.integers()))
def test_elements_preserved(arr):
    """Output must contain exactly the same elements as input."""
    result = my_custom_sort(arr.copy())
    # Works for duplicates too!
    assert sorted(result) == sorted(arr)
 
 
@given(st.lists(st.integers()))
def test_matches_library_sort(arr):
    """Output must match the standard library sort."""
    result = my_custom_sort(arr.copy())
    expected = sorted(arr)
    assert result == expected
 
 
@given(st.lists(st.tuples(st.integers(max_value=10), st.text(max_size=5))))
def test_stability(data):
    """For stable sort: equal elements preserve original order."""
    indexed = [(idx, item) for idx, item in enumerate(data)]
    result = my_custom_sort(indexed.copy(), key=lambda x: x[1][0])
    
    # Check stability: for equal primary keys, indices should be ascending
    for i in range(len(result) - 1):
        if result[i][1][0] == result[i+1][1][0]:  # Same primary key
            assert result[i][0] < result[i+1][0], "Stability violated"
 
 
# === EDGE CASE TESTS ===
 
def test_empty_array():
    assert my_custom_sort([]) == []
 
def test_single_element():
    assert my_custom_sort([42]) == [42]
 
def test_all_equal():
    assert my_custom_sort([5, 5, 5, 5]) == [5, 5, 5, 5]
 
def test_already_sorted():
    assert my_custom_sort([1, 2, 3, 4, 5]) == [1, 2, 3, 4, 5]
 
def test_reverse_sorted():
    assert my_custom_sort([5, 4, 3, 2, 1]) == [1, 2, 3, 4, 5]
 
def test_duplicates_at_edges():
    assert my_custom_sort([1, 1, 3, 4, 4]) == [1, 1, 3, 4, 4]
 
def test_large_array():
    arr = [random.randint(-10000, 10000) for _ in range(100000)]
    assert my_custom_sort(arr.copy()) == sorted(arr)
 
 
# Run with: pytest --hypothesis-show-statistics
# This will run thousands of random test cases