进阶与线性排序

def heap_sort(arr: list[int]) -> list[int]:
    """
    堆排序
    
    Time:  O(n log n) 最坏保证
    Space: O(1)
    Stable: False
    """
    n = len(arr)
    
    def sift_down(start: int, end: int) -> None:
        """在 arr[start..end] 范围内下沉根节点"""
        root = start
        while True:
            child = 2 * root + 1
            if child > end:
                break
            # 选较大子节点
            if child + 1 <= end and arr[child + 1] > arr[child]:
                child += 1
            if arr[root] < arr[child]:
                arr[root], arr[child] = arr[child], arr[root]
                root = child
            else:
                break
    
    # Phase 1: Floyd 自底向上建堆 O(n)
    for i in range(n // 2 - 1, -1, -1):
        sift_down(i, n - 1)
    
    # Phase 2: 反复取出堆顶放到末尾
    for i in range(n - 1, 0, -1):
        arr[0], arr[i] = arr[i], arr[0]
        sift_down(0, i - 1)
    
    return arr

增量序列 {5, 3, 1}，数组 [13, 14, 94, 33, 82, 25, 59, 94, 65, 23, 45, 27, 73, 25, 39, 10]

gap=5 分组（5 个子序列）：
  13 25 45 10  -> 10 13 25 45
  14 59 27     -> 14 27 59
  94 94 73     -> 73 94 94
  33 65 25     -> 25 33 65
  82 23 39     -> 23 39 82
  合并后: [10, 14, 73, 25, 23, 13, 27, 94, 33, 39, 25, 59, 94, 65, 82, 45]

gap=3 分组（3 个子序列），各组插入排序
gap=1 最终插入排序

def shell_sort(arr: list[int], gaps: str = "ciura") -> list[int]:
    """
    希尔排序
    
    Args:
        arr: 待排序数组
        gaps: 增量序列名，可选 "shell" "knuth" "sedgewick" "ciura"
        
    Returns:
        排序后的数组
        
    Time:  依赖增量序列，Ciura 实测最快约 O(n^{1.3})
    Space: O(1)
    Stable: False
    """
    n = len(arr)
    gap_sequences = {
        "shell":     [n // 2, n // 4, ..., 1],  # 原始版本，O(n^{3/2})
        "knuth":     [1, 4, 13, 40, 121, 364, 1093, ...],  # 3h+1
        "sedgewick": [1, 5, 19, 41, 109, 209, 505, 929, 2161, ...],
        "ciura":     [1, 4, 10, 23, 57, 132, 301, 701, 1750],  # 实测最优
    }
    
    if gaps == "shell":
        gap_list = []
        g = n // 2
        while g > 0:
            gap_list.append(g)
            g //= 2
    elif gaps == "knuth":
        gap_list = []
        g = 1
        while g < n:
            gap_list.append(g)
            g = 3 * g + 1
        gap_list.reverse()
    elif gaps == "sedgewick":
        gap_list = []
        k = 0
        while True:
            # Sedgewick 1986: 4^k + 3*2^(k-1) + 1
            if k == 0:
                g = 1
            else:
                g = 4 ** k + 3 * 2 ** (k - 1) + 1
            if g >= n:
                break
            gap_list.append(g)
            k += 1
        gap_list.reverse()
    else:  # ciura
        gap_list = [g for g in [1750, 701, 301, 132, 57, 23, 10, 4, 1] if g < n]
        if not gap_list:
            gap_list = [1]
    
    for gap in gap_list:
        # 对每个子序列做插入排序
        for i in range(gap, n):
            key = arr[i]
            j = i - gap
            while j >= 0 and arr[j] > key:
                arr[j + gap] = arr[j]
                j -= gap
            arr[j + gap] = key
    return arr

def counting_sort(arr: list[int], max_val: int = None) -> list[int]:
    """
    计数排序
    
    Args:
        arr: 待排序数组（非负整数）
        max_val: 元素最大值，默认从 arr 推断
        
    Time:  O(n + k)，k 为值域大小
    Space: O(k)
    Stable: True（按累加计数反向填充）
    """
    if not arr:
        return arr
    if max_val is None:
        max_val = max(arr)
    
    # 统计每个值出现次数
    count = [0] * (max_val + 1)
    for x in arr:
        count[x] += 1
    
    # 累加（确定每个值的结束位置）
    for i in range(1, max_val + 1):
        count[i] += count[i - 1]
    
    # 反向填充保证稳定性
    result = [0] * len(arr)
    for x in reversed(arr):
        count[x] -= 1
        result[count[x]] = x
    
    return result

if __name__ == "__main__":
    ages = [23, 45, 18, 23, 67, 18, 34, 23]
    print(counting_sort(ages, max_val=150))

[18, 18, 23, 23, 23, 34, 45, 67]

def radix_sort_lsd(arr: list[int]) -> list[int]:
    """
    LSD 基数排序（从最低位到最高位）
    
    Time:  O(d(n+k))，d 为位数，k 为基数
    Space: O(n+k)
    Stable: True
    """
    if not arr:
        return arr
    max_val = max(arr)
    exp = 1
    while max_val // exp > 0:
        arr = _counting_sort_by_digit(arr, exp)
        exp *= 10
    return arr

def _counting_sort_by_digit(arr: list[int], exp: int) -> list[int]:
    """按某一位做计数排序"""
    n = len(arr)
    count = [0] * 10
    output = [0] * n
    
    for x in arr:
        digit = (x // exp) % 10
        count[digit] += 1
    
    for i in range(1, 10):
        count[i] += count[i - 1]
    
    # 反向填充保证稳定性
    for x in reversed(arr):
        digit = (x // exp) % 10
        count[digit] -= 1
        output[count[digit]] = x
    
    return output

def radix_sort_msd(arr: list[str], digit: int = 0) -> list[str]:
    """
    MSD 基数排序（从最高位到最低位，用于字符串）
    
    适合变长字符串排序。
    """
    if len(arr) <= 1:
        return arr
    
    buckets = [[] for _ in range(256)]
    finished = []
    for s in arr:
        if digit < len(s):
            buckets[ord(s[digit])].append(s)
        else:
            finished.append(s)
    
    result = finished
    for b in buckets:
        if b:
            result.extend(radix_sort_msd(b, digit + 1))
    return result

if __name__ == "__main__":
    phones = [13812345678, 15987654321, 18600000000, 13800000000]
    print(radix_sort_lsd(phones))

[13800000000, 13812345678, 15987654321, 18600000000]

def bucket_sort(arr: list[float], bucket_count: int = 10) -> list[float]:
    """
    桶排序
    
    Args:
        arr: 待排序数组（浮点数 [0, 1)）
        bucket_count: 桶数
        
    Time:  平均 O(n+k)，最坏 O(n^2)（所有元素入同桶）
    Space: O(n+k)
    Stable: True（若桶内排序稳定）
    """
    if not arr:
        return arr
    
    buckets = [[] for _ in range(bucket_count)]
    for x in arr:
        idx = min(int(x * bucket_count), bucket_count - 1)
        buckets[idx].append(x)
    
    for b in buckets:
        b.sort()  # 桶内排序（用稳定排序）
    
    result = []
    for b in buckets:
        result.extend(b)
    return result

if __name__ == "__main__":
    samples = [0.42, 0.32, 0.23, 0.52, 0.25, 0.47, 0.51]
    print(bucket_sort(samples, bucket_count=10))

[0.23, 0.25, 0.32, 0.42, 0.47, 0.51, 0.52]

import math

def introsort(arr: list[int]) -> list[int]:
    """
    内省排序（introsort）
    
    Musser 1997，C++ std::sort 的核心实现。
    
    Time:  O(n log n) 最坏保证
    Space: O(log n)
    Stable: False
    """
    max_depth = 2 * int(math.log2(len(arr))) if arr else 0
    _introsort_helper(arr, 0, len(arr) - 1, max_depth)
    # 最后做一次插入排序清理小段
    _insertion_sort_range(arr, 0, len(arr) - 1)
    return arr

def _introsort_helper(arr: list[int], low: int, high: int, depth: int) -> None:
    while high - low > 16:
        if depth == 0:
            # 切换到堆排
            _heapsort_range(arr, low, high)
            return
        depth -= 1
        p = _partition_median_three(arr, low, high)
        _introsort_helper(arr, low, p - 1, depth)
        low = p + 1  # 尾递归优化

def _partition_median_three(arr: list[int], low: int, high: int) -> int:
    """三数取中选主元"""
    mid = low + (high - low) // 2
    # 排序 arr[low], arr[mid], arr[high]，取中位数到 high-1
    if arr[low] > arr[mid]:
        arr[low], arr[mid] = arr[mid], arr[low]
    if arr[low] > arr[high]:
        arr[low], arr[high] = arr[high], arr[low]
    if arr[mid] > arr[high]:
        arr[mid], arr[high] = arr[high], arr[mid]
    # 主元放到 high
    arr[mid], arr[high] = arr[high], arr[mid]
    pivot = arr[high]
    i = low - 1
    for j in range(low, high):
        if arr[j] <= pivot:
            i += 1
            arr[i], arr[j] = arr[j], arr[i]
    arr[i + 1], arr[high] = arr[high], arr[i + 1]
    return i + 1

def _heapsort_range(arr: list[int], low: int, high: int) -> None:
    """对 arr[low..high] 做堆排"""
    n = high - low + 1
    for i in range(n // 2 - 1, -1, -1):
        _sift_down(arr, low, i, n)
    for i in range(n - 1, 0, -1):
        arr[low], arr[low + i] = arr[low + i], arr[low]
        _sift_down(arr, low, 0, i)

def _sift_down(arr: list[int], base: int, start: int, size: int) -> None:
    root = start
    while True:
        child = 2 * root + 1
        if child >= size:
            break
        if child + 1 < size and arr[base + child + 1] > arr[base + child]:
            child += 1
        if arr[base + root] < arr[base + child]:
            arr[base + root], arr[base + child] = arr[base + child], arr[base + root]
            root = child
        else:
            break

def _insertion_sort_range(arr: list[int], low: int, high: int) -> None:
    for i in range(low + 1, high + 1):
        key = arr[i]
        j = i - 1
        while j >= low and arr[j] > key:
            arr[j + 1] = arr[j]
            j -= 1
        arr[j + 1] = key

def timsort_simple(arr: list[int]) -> list[int]:
    """
    Timsort 简化版（用于教学）
    
    真实 Timsort 实现复杂得多，参见 CPython listobject.txt
    """
    n = len(arr)
    if n < 64:
        return binary_insertion_sort(arr)
    
    min_run = _compute_minrun(n)
    runs = []
    i = 0
    while i < n:
        # 检测自然 run
        run_end = _find_run(arr, i)
        # 若 run 太短，用二分插入扩展到 min_run
        if run_end - i + 1 < min_run:
            run_end = min(i + min_run - 1, n - 1)
            _binary_insertion_range(arr, i, run_end)
        runs.append((i, run_end))
        i = run_end + 1
        # 维护归并栈不变式
        _merge_collapse(arr, runs)
    
    # 合并所有剩余 run
    _merge_force_collapse(arr, runs)
    return arr

def _compute_minrun(n: int) -> int:
    """计算 minrun：取 6 位使 n/minrun 接近 2^k"""
    r = 0
    while n >= 32:
        r |= n & 1
        n >>= 1
    return n + r

def _find_run(arr: list[int], start: int) -> int:
    """检测自然升序或降序 run"""
    if start == len(arr) - 1:
        return start
    end = start + 1
    if arr[end] >= arr[start]:
        # 升序
        while end + 1 < len(arr) and arr[end + 1] >= arr[end]:
            end += 1
    else:
        # 降序，反转
        while end + 1 < len(arr) and arr[end + 1] < arr[end]:
            end += 1
        arr[start:end + 1] = arr[start:end + 1][::-1]
    return end

def _merge_collapse(arr: list[int], runs: list[tuple[int, int]]) -> None:
    """维护栈顶三 run 不变式"""
    while len(runs) > 1:
        n = len(runs)
        if n >= 3 and (runs[n - 3][1] - runs[n - 3][0] + 1) <= \
                      (runs[n - 2][1] - runs[n - 2][0] + 1) + \
                      (runs[n - 1][1] - runs[n - 1][0] + 1):
            if (runs[n - 3][1] - runs[n - 3][0] + 1) < (runs[n - 1][1] - runs[n - 1][0] + 1):
                _merge_at(arr, runs, n - 3)
            else:
                _merge_at(arr, runs, n - 2)
        elif (runs[n - 2][1] - runs[n - 2][0] + 1) <= (runs[n - 1][1] - runs[n - 1][0] + 1):
            _merge_at(arr, runs, n - 2)
        else:
            break

def _merge_at(arr: list[int], runs: list[tuple[int, int]], idx: int) -> None:
    """合并 runs[idx] 和 runs[idx+1]"""
    left_start, left_end = runs[idx]
    right_start, right_end = runs[idx + 1]
    merged = _merge_gallop(arr[left_start:left_end + 1], arr[right_start:right_end + 1])
    arr[left_start:right_end + 1] = merged
    runs[idx] = (left_start, right_end)
    runs.pop(idx + 1)

def _merge_gallop(left: list[int], right: list[int]) -> list[int]:
    """合并（含 gallop 优化，简化版省略 gallop）"""
    return _merge(left, right)

def _merge_force_collapse(arr: list[int], runs: list[tuple[int, int]]) -> None:
    while len(runs) > 1:
        _merge_at(arr, runs, len(runs) - 2)

def _binary_insertion_range(arr: list[int], low: int, high: int) -> None:
    for i in range(low + 1, high + 1):
        key = arr[i]
        left, right = low, i
        while left < right:
            mid = (left + right) // 2
            if arr[mid] <= key:
                left = mid + 1
            else:
                right = mid
        for j in range(i, left, -1):
            arr[j] = arr[j - 1]
        arr[left] = key

def binary_insertion_sort(arr: list[int]) -> list[int]:
    _binary_insertion_range(arr, 0, len(arr) - 1)
    return arr

# 方案 A：introsort（推荐）
import math
import random

def sortArray(nums: list[int]) -> list[int]:
    """LeetCode 912 - 排序数组（introsort 实现）"""
    if len(nums) <= 1:
        return nums
    max_depth = 2 * int(math.log2(len(nums)))
    _introsort(nums, 0, len(nums) - 1, max_depth)
    return nums

def _introsort(arr, low, high, depth):
    while high - low > 16:
        if depth == 0:
            _heapsort_range(arr, low, high)
            return
        depth -= 1
        # 三数取中
        mid = (low + high) // 2
        if arr[mid] < arr[low]:
            arr[low], arr[mid] = arr[mid], arr[low]
        if arr[high] < arr[low]:
            arr[low], arr[high] = arr[high], arr[low]
        if arr[mid] < arr[high]:
            arr[mid], arr[high] = arr[high], arr[mid]
        pivot = arr[high]
        # 三路分区
        lt, i, gt = low, low, high
        while i <= gt:
            if arr[i] < pivot:
                arr[lt], arr[i] = arr[i], arr[lt]
                lt += 1; i += 1
            elif arr[i] > pivot:
                arr[i], arr[gt] = arr[gt], arr[i]
                gt -= 1
            else:
                i += 1
        _introsort(arr, low, lt - 1, depth)
        low = gt + 1  # 尾递归优化
    _insertion_sort_range(arr, low, high)

# 方案 A：最小堆
import heapq

def findKthLargest_heap(nums: list[int], k: int) -> int:
    """最小堆维护前 K 大元素，O(n log k)"""
    heap = nums[:k]
    heapq.heapify(heap)
    for x in nums[k:]:
        if x > heap[0]:
            heapq.heapreplace(heap, x)
    return heap[0]

# 方案 B：Quickselect（推荐，平均 O(n)）
import random

def findKthLargest_quickselect(nums: list[int], k: int) -> int:
    """Quickselect 选择第 K 大，平均 O(n)"""
    target = len(nums) - k  # 第 K 大 = 升序第 n-k 位
    
    def partition(low, high):
        # 随机化主元
        rand_idx = random.randint(low, high)
        nums[rand_idx], nums[high] = nums[high], nums[rand_idx]
        pivot = nums[high]
        i = low
        for j in range(low, high):
            if nums[j] <= pivot:
                nums[i], nums[j] = nums[j], nums[i]
                i += 1
        nums[i], nums[high] = nums[high], nums[i]
        return i
    
    def quickselect(low, high):
        if low == high:
            return nums[low]
        p = partition(low, high)
        if p == target:
            return nums[p]
        elif p < target:
            return quickselect(p + 1, high)
        else:
            return quickselect(low, p - 1)
    
    return quickselect(0, len(nums) - 1)

// CPython listsort 简化结构（listobject.c）
typedef struct {
    PyObject **ob_item;   // 元素指针数组
    Py_ssize_t allocated; // 容量
    Py_ssize_t used;      // 已用
    // Timsort 内部状态
    MergeState ms;        // 归并栈与临时缓冲
} PyListObject;

typedef struct {
    Py_ssize_t n;          // 栈中 run 数
    struct {
        Py_ssize_t base;   // run 起点
        Py_ssize_t len;    // run 长度
    } pending[MAX_MERGE_PENDING];
    PyObject **temp;       // 临时数组（gallop 缓冲）
    Py_ssize_t alloced;
    Py_ssize_t min_gallop; // gallop 阈值（自适应）
} MergeState;

// JDK 21 DualPivotQuicksort 简化逻辑
public static void sort(int[] a) {
    if (a.length < 286) {
        // 小数组：Dual-Pivot Quicksort
        dualPivotQuicksort(a, 0, a.length - 1, 3);
    } else {
        // 大数组：检测已序性，否则归并排序
        int[] run = new int[68];
        int count = buildRuns(a, run);
        if (count == 1) return;  // 已序
        mergeSort(a, run, count);
    }
}

private static void dualPivotQuicksort(int[] a, int left, int right, int leftmost) {
    int length = right - left + 1;
    // 极小数组用插入排序
    if (length < 47) {
        insertionSort(a, left, right);
        return;
    }
    // 双轴选择：五数取中（e1, e2, e3, e4, e5）
    int pivot1, pivot2;
    // ... 双轴分区
}

// libstdc++ stl_algo.h 简化逻辑
template <typename RandomAccessIterator, typename Compare>
void __sort(RandomAccessIterator first, RandomAccessIterator last, Compare comp) {
    if (last - first > 16) {
        // 内省排序：递归深度限制
        __introsort_loop(first, last, __lg(last - first) * 2, comp);
    }
    // 最终插入排序收尾
    __final_insertion_sort(first, last, comp);
}

template <typename RandomAccessIterator, typename Size, typename Compare>
void __introsort_loop(RandomAccessIterator first, RandomAccessIterator last,
                      Size depth_limit, Compare comp) {
    while (last - first > 16) {
        if (depth_limit == 0) {
            // 切换堆排
            std::__partial_sort(first, last, last, comp);
            return;
        }
        --depth_limit;
        // 三数取中
        RandomAccessIterator cut = __unguarded_partition(
            first, last, __median(*first, *(first + (last-first)/2), *(last-1), comp), comp);
        __introsort_loop(cut, last, depth_limit, comp);  // 递归右半
        last = cut;  // 尾递归优化左半
    }
}

阶段 1（初始排序）：
  - 读取 N 个内存页（buffer_pool_size）
  - 每批内排序（快排或 introsort）
  - 写回磁盘形成 sorted run

阶段 2（多路归并）：
  - 每路 run 读 1 页
  - k 路归并（k = buffer_pool_size / 2）
  - 路数过多时分级归并

// 简化逻辑
void tuplesort_performsort(Tuplesortstate *state) {
    if (state->mem_allowed) {
        // 内存够：quicksort
        tuplesort_sort_inmem(state);
    } else {
        // 超内存：外排序
        // 1. 初始运行
        tuplesort_inittapes(state);
        // 2. 替换选择生成初始 run
        tuplesort_batchinsert(state);
        // 3. 多路归并
        tuplesort_mergeonerun(state);
    }
}

阶段 1（采样）：
  - 预扫描 1MB 数据采样
  - 计算 N-1 个分位点（N 为 reducer 数）

阶段 2（Map）：
  - 每个 mapper 读取本地数据块
  - 根据分位点附加分区号 (key, partition)
  - 输出 (partition, key, value)

阶段 3（Shuffle）：
  - 按 partition 路由到 reducer
  - 同 partition 内有序（Map 端排序 + Reduce 端合并）

阶段 4（Reduce）：
  - 每个 reducer 对本 partition 内数据排序（introsort）
  - 输出到 HDFS

// lib/sort.c
void sort(void *base, size_t num, size_t size,
          int (*cmp_func)(const void *, const void *, void *),
          void *swap_func, void *priv);

# LSD：从低位到高位，每轮稳定排序
def radix_lsd(arr):
    max_val = max(arr)
    exp = 1
    while max_val // exp > 0:
        counting_sort_by_digit(arr, exp)  # 必须稳定！
        exp *= 10

# MSD：从高位到低位，递归分桶
def radix_msd(arr, digit):
    if digit < 0 or len(arr) <= 1:
        return
    buckets = [[] for _ in range(10)]
    for x in arr:
        buckets[(x // 10**digit) % 10].append(x)
    for b in buckets:
        radix_msd(b, digit - 1)
    arr[:] = [x for b in buckets for x in b]