Event-Based Sweeping — Interval Scheduling… | OneNoughtOne

0/86

Data Structures & AlgorithmsInterval Scheduling Patterns

Interval Scheduling Patterns

LevelIntermediate

Duration75 mins

TopicInterval Scheduling Patterns

4 / 4

Event-Based Sweeping

The Sweep Line Paradigm

Throughout this module, we've repeatedly used a powerful technique: sorting events by time and processing them in order. This is the sweep line (or scan line) paradigm—one of the most elegant and widely applicable algorithmic techniques in computer science. The sweep line transforms multi-dimensional problems into simpler one-dimensional processing. Imagine a vertical line sweeping across a plane from left to right. As it moves, it encounters "events" (interval starts, ends, points, line segment intersections). At each event, we update some state structure. This event-driven processing often turns O(n²) brute force into O(n log n) elegance. In this page, we'll formalize the sweep line paradigm, see it as the unifying framework behind interval problems, extend it to more complex scenarios, and explore advanced applications beyond simple counting.

What You Will Learn

By the end of this page, you will understand the sweep line paradigm formally, recognize when sweep line applies, implement sweep algorithms with various state structures, handle event types beyond simple starts/ends, apply sweep to geometric problems, and connect sweep line to broader algorithm design patterns.

The Sweep Line Framework

The sweep line paradigm has a consistent structure across applications: 1. Event Generation Convert input objects into events with positions (typically x-coordinates or times). Each event has: A position (where the sweep line encounters it) A type (start, end, point, intersection, etc.) Associated data (which interval, what properties) 2. Event Sorting Sort events by position. Tie-breaking rules handle simultaneous events: Often: process "end" before "start" at same position Sometimes: other orders based on problem semantics 3. State Structure Maintain a data structure representing "what's currently active" or "what's currently known" at the sweep position. This might be: A counter (for counting overlaps) A set or multiset (for tracking active intervals) A balanced BST (for ordered active segments) A segment tree or BIT (for range queries) 4. Event Processing Sweep through events in sorted order. At each event: Update the state structure (add/remove intervals, update counts) Query the state if needed (max overlap, active intervals containing a point) Accumulate answers or detect conditions 5. Result Extraction After processing all events (or during), extract the answer from accumulated state or queries.

Sweep Line Components Across Problems
Problem	Events	State Structure	Query/Update
Meeting Rooms II	Start (+1), End (-1)	Counter	Track max counter value
Interval Stabbing	Interval as [start, end]	Current interval end	Place point when gap detected
Range Coverage	Interval bounds	Current coverage reach	Extend reach greedily
Rectangle Union Area	Left/right edges	Multiset of active y-ranges	Compute total y-coverage
Line Segment Intersection	Segment endpoints	Balanced BST of active segments	Check neighbors for intersection

The Power of Ordering

The key insight of sweep line is that ordering by one dimension converts a 2D problem into a 1D process. At each position, we only need to consider 'active' elements—those overlapping the current sweep position. This reduction is what makes sweep algorithms efficient.

Event Types and Priority Ordering

Event design is crucial for correctness. Different problems require different event types and orderings. Common Event Types: INTERVAL_START — An interval begins; add it to active set INTERVAL_END — An interval ends; remove from active set POINT — A query or marker point; query active intervals SEGMENT_START — A line segment begins (for geometric problems) SEGMENT_END — A line segment ends INTERSECTION — Two segments cross (computed dynamically) Priority Ordering (Tie-Breaking): When multiple events share the same position, the processing order matters: Example 1: Counting Overlap at a Point To correctly count intervals containing a query point p when intervals [a, b] have b = p and new intervals [p, c] start: If we count [a, p] as containing p, process starts before ends If [a, p) excludes p, process ends before starts Example 2: Platform Sharing If train A departs at 10:00 and train B arrives at 10:00, they can share a platform. Process departures before arrivals at the same time.

event_design.py
Python
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
from enum import Enum
from dataclasses import dataclass
from typing import Any
import heapq
 
class EventType(Enum):
    """Event types with natural priority ordering."""
    # Lower value = higher priority (processed first at same position)
    END = 0       # Process ends first (interval closes)
    POINT = 1     # Then query points
    START = 2     # Then starts (interval opens)
 
 
@dataclass(order=True)
class Event:
    """
    An event for sweep line processing.
    
    Ordering: by position first, then by type priority.
    Using dataclass(order=True) auto-generates comparison methods.
    """
    position: float
    event_type: EventType
    data: Any = None  # Excluded from ordering
    
    def __post_init__(self):
        # For proper ordering, store type's value
        self._type_priority = self.event_type.value
 
 
def create_interval_events(intervals: list[tuple[int, int]]) -> list[Event]:
    """
    Convert intervals to events.
    
    Each interval [start, end] becomes two events:
    - START at position 'start'
    - END at position 'end'
    """
    events = []
    for i, (start, end) in enumerate(intervals):
        events.append(Event(start, EventType.START, data={'interval_id': i}))
        events.append(Event(end, EventType.END, data={'interval_id': i}))
    return sorted(events, key=lambda e: (e.position, e.event_type.value))
 
 
def create_point_query_events(
    intervals: list[tuple[int, int]], 
    query_points: list[int]
) -> list[Event]:
    """
    Create events for intervals and query points.
    
    Allows answering: 'How many intervals contain point p?' for multiple p.
    """
    events = []
    
    for i, (start, end) in enumerate(intervals):
        events.append(Event(start, EventType.START, data={'interval_id': i}))
        events.append(Event(end, EventType.END, data={'interval_id': i}))
    
    for j, point in enumerate(query_points):
        events.append(Event(point, EventType.POINT, data={'query_id': j}))
    
    # Sort by position, then by type priority
    return sorted(events, key=lambda e: (e.position, e.event_type.value))
 
 
# Example: Count intervals containing each query point
def intervals_containing_points(
    intervals: list[tuple[int, int]], 
    query_points: list[int]
) -> list[int]:
    """
    For each query point, count how many intervals contain it.
    
    Uses sweep line with event priority:
    - END events first (close intervals before checking)
    - POINT events next (query current count)
    - START events last (open intervals after checking)
    
    Wait, actually for "contains" we usually want START before POINT before END...
    This depends on open vs closed intervals. Let's assume [start, end).
    
    For [start, end) (closed-start, open-end):
    - START before POINT (at same position, point is inside)
    - POINT before END (at same position, point is NOT inside [a, p))
    
    Adjusting EventType priorities accordingly.
    """
    # Redefine for this variant
    events = []
    
    PRIORITY_START = 0
    PRIORITY_POINT = 1
    PRIORITY_END = 2
    
    for i, (start, end) in enumerate(intervals):
        events.append((start, PRIORITY_START, 'START', i))
        events.append((end, PRIORITY_END, 'END', i))
    
    for j, point in enumerate(query_points):
        events.append((point, PRIORITY_POINT, 'POINT', j))
    
    events.sort()  # Sort by (position, priority)
    
    active_count = 0
    results = [0] * len(query_points)
    
    for pos, priority, event_type, idx in events:
        if event_type == 'START':
            active_count += 1
        elif event_type == 'END':
            active_count -= 1
        elif event_type == 'POINT':
            results[idx] = active_count
    
    return results
 
 
# Example usage
intervals = [(1, 5), (2, 6), (4, 8)]
points = [0, 3, 5, 7]
 
result = intervals_containing_points(intervals, points)
print(f"Points: {points}")
print(f"Counts: {result}")  # [0, 2, 1, 1] for [1,5), [2,6), [4,8)

Priority Order Depends on Semantics

There's no universal 'correct' priority order. It depends on: (1) Are intervals open or closed? (2) What exactly are you computing? (3) What's the natural interpretation? Always reason through a small example with ties to verify your ordering.

State Structures for Sweep Line

The power of sweep line comes from maintaining efficient state as we process events. Different problems require different state structures: 1. Counter / Variable Simplest case Tracks: count of active intervals, sum of values, minimum/maximum Update: O(1) Query: O(1) Example: Meeting Rooms II 2. Set / Multiset Tracks: which intervals are currently active (by ID or full data) Update: O(log n) for balanced tree-based set Query: O(1) for size, O(log n) for membership Example: Finding intervals containing a point 3. Ordered Set / Balanced BST Tracks: active elements in sorted order (by y-coordinate, for example) Update: O(log n) Query: O(log n) for predecessor/successor, min/max Example: Line segment intersection (Bentley-Ottmann) 4. Segment Tree / BIT Tracks: range information (sums, maxes) over an axis Update: O(log n) Query: O(log n) for range queries Example: Rectangle area union with continuous y-ranges

state_structures.py
Python
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
from collections import defaultdict
from sortedcontainers import SortedList  # pip install sortedcontainers
 
# ==================================================
# Example 1: Simple Counter State
# ==================================================
 
def max_overlap_counter(intervals: list[tuple[int, int]]) -> int:
    """
    Maximum overlap using counter state.
    State: single integer 'count'
    """
    events = []
    for start, end in intervals:
        events.append((start, 1))   # +1 at start
        events.append((end, -1))    # -1 at end
    
    events.sort(key=lambda x: (x[0], x[1]))  # End before start at ties
    
    count = 0
    max_count = 0
    
    for _, delta in events:
        count += delta
        max_count = max(max_count, count)
    
    return max_count
 
 
# ==================================================
# Example 2: Set State - Track Active Interval IDs
# ==================================================
 
def intervals_at_each_point(
    intervals: list[tuple[int, int]], 
    points: list[int]
) -> dict[int, list[int]]:
    """
    For each query point, return LIST of interval indices containing it.
    State: set of active interval IDs
    """
    events = []
    
    for i, (start, end) in enumerate(intervals):
        events.append((start, 0, 'START', i))  # 0 = process first
        events.append((end, 2, 'END', i))       # 2 = process last
    
    for j, p in enumerate(points):
        events.append((p, 1, 'QUERY', j))       # 1 = process middle
    
    events.sort()
    
    active = set()  # State: set of active interval IDs
    results = {}
    
    for pos, _, event_type, idx in events:
        if event_type == 'START':
            active.add(idx)
        elif event_type == 'END':
            active.discard(idx)
        elif event_type == 'QUERY':
            results[points[idx]] = list(active)
    
    return results
 
 
# ==================================================
# Example 3: Ordered State - Skyline Problem
# ==================================================
 
def get_skyline(buildings: list[list[int]]) -> list[list[int]]:
    """
    The Skyline Problem: given buildings as [left, right, height],
    return the skyline silhouette as list of [x, height] key points.
    
    State: multiset of active building heights (ordered)
    At each x, the skyline height = max of active heights (or 0 if none)
    
    Uses SortedList for O(log n) insertion/removal of heights.
    """
    # Events: (x, type, height)
    # type: 0 = building starts (entering from left)
    #       1 = building ends (exiting to right)
    events = []
    
    for left, right, height in buildings:
        events.append((left, 0, height))   # Start: add height
        events.append((right, 1, height))  # End: remove height
    
    # Sort: by x, then starts before ends, then taller starts first
    events.sort(key=lambda e: (e[0], e[1], -e[2] if e[1] == 0 else e[2]))
    
    # State: sorted list of active heights
    # We use a sorted list to easily get max
    active_heights = SortedList([0])  # 0 as sentinel for ground level
    
    result = []
    prev_max_height = 0
    
    for x, event_type, height in events:
        if event_type == 0:  # Building starts
            active_heights.add(height)
        else:  # Building ends
            active_heights.remove(height)
        
        # Current max height
        current_max = active_heights[-1]
        
        # If max height changed, record key point
        if current_max != prev_max_height:
            result.append([x, current_max])
            prev_max_height = current_max
    
    return result
 
 
# ==================================================
# Example 4: Counter per Y-coordinate (for area computation)
# ==================================================
 
def rectangles_area_union(rectangles: list[list[int]]) -> int:
    """
    Compute total area of union of axis-aligned rectangles.
    
    Sweep along x-axis, maintaining active y-ranges.
    At each x-event, compute change in covered area.
    
    Simplified version using coordinate compression.
    """
    # Collect all y-coordinates for compression
    y_coords = set()
    events = []
    
    for x1, y1, x2, y2 in rectangles:
        y_coords.add(y1)
        y_coords.add(y2)
        events.append((x1, 0, y1, y2))  # LEFT edge (add range)
        events.append((x2, 1, y1, y2))  # RIGHT edge (remove range)
    
    # Coordinate compression
    y_list = sorted(y_coords)
    y_to_idx = {y: i for i, y in enumerate(y_list)}
    
    # Count array: count[i] = number of active rectangles covering [y_list[i], y_list[i+1])
    count = [0] * len(y_list)
    
    events.sort()
    
    total_area = 0
    prev_x = events[0][0] if events else 0
    
    for x, event_type, y1, y2 in events:
        # Compute covered y-length
        covered_y = 0
        for i in range(len(y_list) - 1):
            if count[i] > 0:
                covered_y += y_list[i + 1] - y_list[i]
        
        # Add area since last x
        total_area += covered_y * (x - prev_x)
        prev_x = x
        
        # Update counts
        idx1, idx2 = y_to_idx[y1], y_to_idx[y2]
        delta = 1 if event_type == 0 else -1
        for i in range(idx1, idx2):
            count[i] += delta
    
    return total_area
 
 
# Example usage
print("Max overlap:", max_overlap_counter([(1, 5), (2, 6), (4, 8)]))
 
intervals = [(0, 10), (3, 7), (5, 15)]
points = [1, 5, 8, 12]
print("Intervals at points:", intervals_at_each_point(intervals, points))
 
buildings = [[2, 9, 10], [3, 7, 15], [5, 12, 12], [15, 20, 10], [19, 24, 8]]
print("Skyline:", get_skyline(buildings))

Choosing the Right State Structure

Counter: when you only need aggregate (count, sum, max). Set: when you need to know which intervals are active. Ordered structure (SortedList, BST): when you need the max/min of active values, or predecessor/successor queries. Segment tree: when you need efficient range queries on the state.

The Skyline Problem: A Classic Sweep Application

The Skyline Problem is a celebrated application of sweep line that appears in coding interviews and demonstrates the paradigm's power. Problem Statement: Given a list of buildings where each building is represented as [left, right, height] : left = x-coordinate of left edge right = x-coordinate of right edge height = building height Return the skyline formed by these buildings as a list of "key points" [x, height] where the height changes. The Sweep Line Approach: Events: Each building creates two events: (left, START, height) — building enters at x = left (right, END, height) — building exits at x = right State: A multiset (sorted list) of active building heights Processing: At each x-position, after updating the state, check if the maximum height changed. If so, record a key point. The Critical Insight: The skyline height at any x equals the maximum height among all buildings covering that x. By maintaining active heights in a max-queryable structure, we efficiently track this as the sweep progresses.

Skyline Problem Example

Computing the city skyline

Input

Buildings: [[2, 9, 10], [3, 7, 15], [5, 12, 12], [15, 20, 10], [19, 24, 8]]

Output

Skyline: [[2, 10], [3, 15], [7, 12], [12, 0], [15, 10], [20, 8], [24, 0]]

Explanation

At x=2, we rise to height 10. At x=3, building of height 15 starts (higher). At x=7, that building ends, dropping to 12. At x=12, last tall building ends, dropping to 0. And so on.

skyline.py
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
import heapq
from collections import defaultdict
 
def get_skyline(buildings: list[list[int]]) -> list[list[int]]:
    """
    Compute the skyline from a list of buildings.
    
    Uses a max-heap to track active building heights.
    Lazy deletion handles building exits.
    
    Time: O(n log n)
    Space: O(n)
    """
    if not buildings:
        return []
    
    # Events: (x, type, height)
    # type: 0 = start (entering), 1 = end (leaving)
    events = []
    
    for left, right, height in buildings:
        events.append((left, 0, height))   # Building starts
        events.append((right, 1, height))  # Building ends
    
    # Sort: by x, then starts before ends, then taller starts first
    # For ends: process shorter heights first (doesn't matter if using lazy deletion)
    events.sort(key=lambda e: (e[0], e[1], -e[2]))
    
    # Max-heap of active heights (use negative for max-heap with heapq)
    # We'll use lazy deletion: track counts of each height
    heap = [0]  # Start with ground level
    height_count = defaultdict(int)  # height -> active count
    height_count[0] = 1  # Ground always active
    
    result = []
    prev_max = 0
    
    i = 0
    while i < len(events):
        curr_x = events[i][0]
        
        # Process all events at the same x
        while i < len(events) and events[i][0] == curr_x:
            x, event_type, height = events[i]
            
            if event_type == 0:  # Start
                heapq.heappush(heap, -height)
                height_count[height] += 1
            else:  # End
                height_count[height] -= 1
            
            i += 1
        
        # Lazy deletion: pop heights that are no longer active
        while heap and height_count[-heap[0]] == 0:
            heapq.heappop(heap)
        
        # Current max height
        curr_max = -heap[0] if heap else 0
        
        # If max changed, record key point
        if curr_max != prev_max:
            result.append([curr_x, curr_max])
            prev_max = curr_max
    
    return result
 
 
# Cleaner version using SortedList (from sortedcontainers)
def get_skyline_sortedlist(buildings: list[list[int]]) -> list[list[int]]:
    """
    Skyline using SortedList for O(log n) max queries.
    Cleaner but requires external library.
    """
    from sortedcontainers import SortedList
    
    events = []
    for left, right, height in buildings:
        events.append((left, -height, 0))   # Start: negative height for sorting
        events.append((right, height, 1))   # End: positive height
    
    # Sort by x, then by value (starts before ends at same x, taller starts first)
    events.sort()
    
    active = SortedList([0])  # Active heights
    result = []
    prev_max = 0
    
    for x, h, event_type in events:
        if event_type == 0:  # Start
            active.add(-h)
        else:  # End
            active.remove(h)
        
        curr_max = active[-1]  # Max is last element in sorted list
        
        if curr_max != prev_max:
            result.append([x, curr_max])
            prev_max = curr_max
    
    return result
 
 
# Test
buildings = [[2, 9, 10], [3, 7, 15], [5, 12, 12], [15, 20, 10], [19, 24, 8]]
print("Skyline:", get_skyline(buildings))

Beyond Intervals: Geometric Sweep

Sweep line extends far beyond 1D intervals into 2D geometry and beyond. Line Segment Intersection (Bentley-Ottmann) Given a set of line segments, find all intersection points. Naive O(n²): Check every pair. Sweep O((n + k) log n): Where k = number of intersections Events: segment start, segment end, intersection (detected dynamically) State: balanced BST of active segments ordered by y-coordinate at current sweep x Key insight: only adjacent segments in the BST can intersect; swap them when they cross Closest Pair of Points Find the two closest points in a 2D plane. Sweep approach: Sort points by x-coordinate Maintain active points within distance d of current x (in a y-ordered structure) Only check active points for closer pairs Prune points that fall too far behind Voronoi Diagrams (Fortune's Algorithm) Construct Voronoi diagram using a sweep line, maintaining a "beach line" of parabolic arcs.

Advanced Sweep Line Applications

•Rectangle intersection detection — Find overlapping rectangle pairs in O((n + k) log n)
•Area of rectangle union — Covered area by overlapping rectangles
•Polygon triangulation — Decompose simple polygon into triangles
•Map overlay — Compute intersection of two planar subdivisions
•Visibility computation — What's visible from a point through obstacles
•Motion planning — Robot path planning using configuration space sweeps

The Sweep Principle Generalizes

Any problem where ordering by one dimension lets you process the remaining dimensions locally is a candidate for sweep line. The key question: 'Can I process objects incrementally as I move through space/time, updating a manageable state structure?'

Implementation Patterns and Best Practices

Let's codify the patterns that make sweep line implementations robust: Pattern 1: Event Class with Natural Ordering @dataclass(order=True) class Event: position: float priority: int # Tie-breaking data: Any = field(compare=False) # Not for ordering Pattern 2: Batch Processing at Same Position When multiple events share a position, process them all before updating the result: i = 0 while i < len(events): curr_pos = events[i].position while i < len(events) and events[i].position == curr_pos: process(events[i]) i += 1 record_result_at(curr_pos) Pattern 3: Lazy Deletion with Heaps When using heaps that don't support arbitrary removal: while heap and is_deleted(heap_top()): heapq.heappop(heap) current_value = heap_top() Pattern 4: Coordinate Compression When coordinates are sparse in a large range: coords = sorted(set(all_coordinates)) coord_to_index = {c: i for i, c in enumerate(coords)} # Now use indices instead of raw coordinates

sweep_patterns.py
Python
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
from dataclasses import dataclass, field
from typing import Any, Callable
from enum import IntEnum
import heapq
 
# =========================================
# Pattern: Generic Sweep Line Framework
# =========================================
 
class EventPriority(IntEnum):
    """Standard event priorities."""
    END = 0
    QUERY = 1  
    START = 2
 
 
@dataclass(order=True)
class SweepEvent:
    """Generic sweep event with natural ordering."""
    position: float
    priority: int  # Lower = processed first at same position
    data: Any = field(compare=False)
 
 
def sweep_line_generic(
    events: list[SweepEvent],
    on_start: Callable[[Any], None],
    on_end: Callable[[Any], None],
    on_query: Callable[[Any], Any],
) -> list:
    """
    Generic sweep line processor.
    
    Separates event generation from processing logic.
    """
    events.sort()
    results = []
    
    for event in events:
        if event.priority == EventPriority.START:
            on_start(event.data)
        elif event.priority == EventPriority.END:
            on_end(event.data)
        elif event.priority == EventPriority.QUERY:
            result = on_query(event.data)
            results.append(result)
    
    return results
 
 
# =========================================
# Example: Using the Framework
# =========================================
 
def count_intervals_at_points(intervals, query_points):
    """
    Use generic framework to count intervals at query points.
    """
    events = []
    
    for i, (start, end) in enumerate(intervals):
        events.append(SweepEvent(start, EventPriority.START, {'id': i}))
        events.append(SweepEvent(end, EventPriority.END, {'id': i}))
    
    for j, p in enumerate(query_points):
        events.append(SweepEvent(p, EventPriority.QUERY, {'point': p, 'idx': j}))
    
    # State
    count = 0
    results = [0] * len(query_points)
    
    def on_start(data):
        nonlocal count
        count += 1
    
    def on_end(data):
        nonlocal count
        count -= 1
    
    def on_query(data):
        results[data['idx']] = count
        return (data['point'], count)
    
    sweep_line_generic(events, on_start, on_end, on_query)
    return results
 
 
# =========================================
# Pattern: Heap with Lazy Deletion
# =========================================
 
class LazyMaxHeap:
    """
    Max-heap supporting lazy deletion.
    
    Items can be 'removed' without actually popping;
    they're skipped when accessed.
    """
    def __init__(self):
        self.heap = []  # Min-heap of negative values
        self.removed = {}  # value -> removal count
    
    def push(self, value):
        heapq.heappush(self.heap, -value)
    
    def remove(self, value):
        """Mark value for removal (lazy)."""
        self.removed[value] = self.removed.get(value, 0) + 1
    
    def _clean_top(self):
        """Remove lazily deleted items from top."""
        while self.heap:
            top = -self.heap[0]
            if self.removed.get(top, 0) > 0:
                heapq.heappop(self.heap)
                self.removed[top] -= 1
            else:
                break
    
    def max(self):
        """Return current maximum."""
        self._clean_top()
        return -self.heap[0] if self.heap else 0
    
    def pop(self):
        """Remove and return maximum."""
        self._clean_top()
        return -heapq.heappop(self.heap) if self.heap else 0
 
 
# Example: Skyline with lazy heap
def skyline_with_lazy_heap(buildings):
    """Skyline using LazyMaxHeap."""
    events = []
    for left, right, height in buildings:
        events.append((left, 0, height))
        events.append((right, 1, height))
    
    events.sort(key=lambda e: (e[0], e[1], -e[2] if e[1] == 0 else e[2]))
    
    heap = LazyMaxHeap()
    heap.push(0)  # Ground level
    
    result = []
    prev_max = 0
    
    i = 0
    while i < len(events):
        curr_x = events[i][0]
        
        while i < len(events) and events[i][0] == curr_x:
            x, event_type, height = events[i]
            if event_type == 0:
                heap.push(height)
            else:
                heap.remove(height)
            i += 1
        
        curr_max = heap.max()
        if curr_max != prev_max:
            result.append([curr_x, curr_max])
            prev_max = curr_max
    
    return result
 
 
# Test
print(count_intervals_at_points([(1, 5), (2, 6), (4, 8)], [0, 3, 5, 7]))
print(skyline_with_lazy_heap([[2, 9, 10], [3, 7, 15], [5, 12, 12]]))

Common Mistakes and Debugging

Sweep line bugs are often subtle. Here are the most common issues: Mistake 1: Wrong Event Priority Your algorithm gives wrong results when events coincide. Debug by: Listing tied events and the expected processing order Checking that your sort key produces this order Mistake 2: Off-By-One in State Update Updating state after recording instead of before, or vice versa. The general rule: Update state for the current event THEN query/record the result Mistake 3: Forgetting Ground Level / Base Case In skyline, forgetting to initialize with height 0 (ground level). The heap should never be truly empty. Mistake 4: Floating Point Comparison When positions are floats, equality checks need tolerance: if abs(event.position - current_position) < 1e-9: Mistake 5: Not Handling Empty State What's the max height when no buildings are active? The minimum overlap when no intervals? Always define the "nothing active" case.

Debugging Checklist

•Print events after sorting — Verify order is correct for tied positions
•Trace state at key positions — Add logging to show state changes
•Test with trivial cases — 0, 1, 2 intervals; no overlap; all overlap
•Test with ties — Events at same position reveal priority bugs
•Draw it out — Sketch intervals on a number line, manually trace the sweep
•Compare with brute force — For small inputs, verify against O(n²) solution

debugging_helper.py
Python
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
def debug_sweep(events, description="Sweep"):
    """
    Print events in processing order for debugging.
    """
    print(f"
=== {description} ===")
    print("Sorted events:")
    for e in sorted(events, key=lambda x: (x[0], x[1])):
        print(f"  Position {e[0]:6.2f}, Priority {e[1]}, Data: {e[2:]}")
    print()
 
 
def verify_with_brute_force(intervals, query_points, sweep_result):
    """
    Verify sweep result against naive O(n*m) computation.
    """
    brute_result = []
    for p in query_points:
        count = sum(1 for s, e in intervals if s <= p < e)
        brute_result.append(count)
    
    if sweep_result == brute_result:
        print("✓ Sweep result matches brute force")
    else:
        print("✗ Mismatch!")
        print(f"  Sweep:  {sweep_result}")
        print(f"  Brute:  {brute_result}")
        for i, p in enumerate(query_points):
            if sweep_result[i] != brute_result[i]:
                print(f"  Point {p}: Sweep={sweep_result[i]}, Brute={brute_result[i]}")
 
 
# Example usage
intervals = [(1, 5), (2, 6), (4, 8)]
points = [0, 1, 2, 3, 4, 5, 6, 7, 8]
 
# Create events for debugging
events = []
for i, (s, e) in enumerate(intervals):
    events.append((s, 0, 'START', i))
    events.append((e, 2, 'END', i))
for j, p in enumerate(points):
    events.append((p, 1, 'QUERY', j))
 
debug_sweep(events, "Interval Count Sweep")
 
# The actual sweep (simplified)
events_sorted = sorted(events)
active = 0
results = [0] * len(points)
 
for pos, priority, event_type, idx in events_sorted:
    if event_type == 'START':
        active += 1
    elif event_type == 'END':
        active -= 1
    elif event_type == 'QUERY':
        results[idx] = active
 
print(f"Sweep results: {results}")
verify_with_brute_force(intervals, points, results)

Data Structures & AlgorithmsInterval Scheduling Patterns

Interval Scheduling Patterns

LevelIntermediate

Duration75 mins

TopicInterval Scheduling Patterns

4 / 4

Event-Based Sweeping

The Sweep Line Paradigm

Throughout this module, we've repeatedly used a powerful technique: sorting events by time and processing them in order. This is the sweep line (or scan line) paradigm—one of the most elegant and widely applicable algorithmic techniques in computer science. The sweep line transforms multi-dimensional problems into simpler one-dimensional processing. Imagine a vertical line sweeping across a plane from left to right. As it moves, it encounters "events" (interval starts, ends, points, line segment intersections). At each event, we update some state structure. This event-driven processing often turns O(n²) brute force into O(n log n) elegance. In this page, we'll formalize the sweep line paradigm, see it as the unifying framework behind interval problems, extend it to more complex scenarios, and explore advanced applications beyond simple counting.

What You Will Learn

By the end of this page, you will understand the sweep line paradigm formally, recognize when sweep line applies, implement sweep algorithms with various state structures, handle event types beyond simple starts/ends, apply sweep to geometric problems, and connect sweep line to broader algorithm design patterns.

The Sweep Line Framework

The sweep line paradigm has a consistent structure across applications: 1. Event Generation Convert input objects into events with positions (typically x-coordinates or times). Each event has: A position (where the sweep line encounters it) A type (start, end, point, intersection, etc.) Associated data (which interval, what properties) 2. Event Sorting Sort events by position. Tie-breaking rules handle simultaneous events: Often: process "end" before "start" at same position Sometimes: other orders based on problem semantics 3. State Structure Maintain a data structure representing "what's currently active" or "what's currently known" at the sweep position. This might be: A counter (for counting overlaps) A set or multiset (for tracking active intervals) A balanced BST (for ordered active segments) A segment tree or BIT (for range queries) 4. Event Processing Sweep through events in sorted order. At each event: Update the state structure (add/remove intervals, update counts) Query the state if needed (max overlap, active intervals containing a point) Accumulate answers or detect conditions 5. Result Extraction After processing all events (or during), extract the answer from accumulated state or queries.

Sweep Line Components Across Problems
Problem	Events	State Structure	Query/Update
Meeting Rooms II	Start (+1), End (-1)	Counter	Track max counter value
Interval Stabbing	Interval as [start, end]	Current interval end	Place point when gap detected
Range Coverage	Interval bounds	Current coverage reach	Extend reach greedily
Rectangle Union Area	Left/right edges	Multiset of active y-ranges	Compute total y-coverage
Line Segment Intersection	Segment endpoints	Balanced BST of active segments	Check neighbors for intersection

The Power of Ordering

The key insight of sweep line is that ordering by one dimension converts a 2D problem into a 1D process. At each position, we only need to consider 'active' elements—those overlapping the current sweep position. This reduction is what makes sweep algorithms efficient.

Event Types and Priority Ordering

Event design is crucial for correctness. Different problems require different event types and orderings. Common Event Types: INTERVAL_START — An interval begins; add it to active set INTERVAL_END — An interval ends; remove from active set POINT — A query or marker point; query active intervals SEGMENT_START — A line segment begins (for geometric problems) SEGMENT_END — A line segment ends INTERSECTION — Two segments cross (computed dynamically) Priority Ordering (Tie-Breaking): When multiple events share the same position, the processing order matters: Example 1: Counting Overlap at a Point To correctly count intervals containing a query point p when intervals [a, b] have b = p and new intervals [p, c] start: If we count [a, p] as containing p, process starts before ends If [a, p) excludes p, process ends before starts Example 2: Platform Sharing If train A departs at 10:00 and train B arrives at 10:00, they can share a platform. Process departures before arrivals at the same time.

event_design.py
Python
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
from enum import Enum
from dataclasses import dataclass
from typing import Any
import heapq
 
class EventType(Enum):
    """Event types with natural priority ordering."""
    # Lower value = higher priority (processed first at same position)
    END = 0       # Process ends first (interval closes)
    POINT = 1     # Then query points
    START = 2     # Then starts (interval opens)
 
 
@dataclass(order=True)
class Event:
    """
    An event for sweep line processing.
    
    Ordering: by position first, then by type priority.
    Using dataclass(order=True) auto-generates comparison methods.
    """
    position: float
    event_type: EventType
    data: Any = None  # Excluded from ordering
    
    def __post_init__(self):
        # For proper ordering, store type's value
        self._type_priority = self.event_type.value
 
 
def create_interval_events(intervals: list[tuple[int, int]]) -> list[Event]:
    """
    Convert intervals to events.
    
    Each interval [start, end] becomes two events:
    - START at position 'start'
    - END at position 'end'
    """
    events = []
    for i, (start, end) in enumerate(intervals):
        events.append(Event(start, EventType.START, data={'interval_id': i}))
        events.append(Event(end, EventType.END, data={'interval_id': i}))
    return sorted(events, key=lambda e: (e.position, e.event_type.value))
 
 
def create_point_query_events(
    intervals: list[tuple[int, int]], 
    query_points: list[int]
) -> list[Event]:
    """
    Create events for intervals and query points.
    
    Allows answering: 'How many intervals contain point p?' for multiple p.
    """
    events = []
    
    for i, (start, end) in enumerate(intervals):
        events.append(Event(start, EventType.START, data={'interval_id': i}))
        events.append(Event(end, EventType.END, data={'interval_id': i}))
    
    for j, point in enumerate(query_points):
        events.append(Event(point, EventType.POINT, data={'query_id': j}))
    
    # Sort by position, then by type priority
    return sorted(events, key=lambda e: (e.position, e.event_type.value))
 
 
# Example: Count intervals containing each query point
def intervals_containing_points(
    intervals: list[tuple[int, int]], 
    query_points: list[int]
) -> list[int]:
    """
    For each query point, count how many intervals contain it.
    
    Uses sweep line with event priority:
    - END events first (close intervals before checking)
    - POINT events next (query current count)
    - START events last (open intervals after checking)
    
    Wait, actually for "contains" we usually want START before POINT before END...
    This depends on open vs closed intervals. Let's assume [start, end).
    
    For [start, end) (closed-start, open-end):
    - START before POINT (at same position, point is inside)
    - POINT before END (at same position, point is NOT inside [a, p))
    
    Adjusting EventType priorities accordingly.
    """
    # Redefine for this variant
    events = []
    
    PRIORITY_START = 0
    PRIORITY_POINT = 1
    PRIORITY_END = 2
    
    for i, (start, end) in enumerate(intervals):
        events.append((start, PRIORITY_START, 'START', i))
        events.append((end, PRIORITY_END, 'END', i))
    
    for j, point in enumerate(query_points):
        events.append((point, PRIORITY_POINT, 'POINT', j))
    
    events.sort()  # Sort by (position, priority)
    
    active_count = 0
    results = [0] * len(query_points)
    
    for pos, priority, event_type, idx in events:
        if event_type == 'START':
            active_count += 1
        elif event_type == 'END':
            active_count -= 1
        elif event_type == 'POINT':
            results[idx] = active_count
    
    return results
 
 
# Example usage
intervals = [(1, 5), (2, 6), (4, 8)]
points = [0, 3, 5, 7]
 
result = intervals_containing_points(intervals, points)
print(f"Points: {points}")
print(f"Counts: {result}")  # [0, 2, 1, 1] for [1,5), [2,6), [4,8)

Priority Order Depends on Semantics

There's no universal 'correct' priority order. It depends on: (1) Are intervals open or closed? (2) What exactly are you computing? (3) What's the natural interpretation? Always reason through a small example with ties to verify your ordering.

State Structures for Sweep Line

The power of sweep line comes from maintaining efficient state as we process events. Different problems require different state structures: 1. Counter / Variable Simplest case Tracks: count of active intervals, sum of values, minimum/maximum Update: O(1) Query: O(1) Example: Meeting Rooms II 2. Set / Multiset Tracks: which intervals are currently active (by ID or full data) Update: O(log n) for balanced tree-based set Query: O(1) for size, O(log n) for membership Example: Finding intervals containing a point 3. Ordered Set / Balanced BST Tracks: active elements in sorted order (by y-coordinate, for example) Update: O(log n) Query: O(log n) for predecessor/successor, min/max Example: Line segment intersection (Bentley-Ottmann) 4. Segment Tree / BIT Tracks: range information (sums, maxes) over an axis Update: O(log n) Query: O(log n) for range queries Example: Rectangle area union with continuous y-ranges

state_structures.py
Python
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
from collections import defaultdict
from sortedcontainers import SortedList  # pip install sortedcontainers
 
# ==================================================
# Example 1: Simple Counter State
# ==================================================
 
def max_overlap_counter(intervals: list[tuple[int, int]]) -> int:
    """
    Maximum overlap using counter state.
    State: single integer 'count'
    """
    events = []
    for start, end in intervals:
        events.append((start, 1))   # +1 at start
        events.append((end, -1))    # -1 at end
    
    events.sort(key=lambda x: (x[0], x[1]))  # End before start at ties
    
    count = 0
    max_count = 0
    
    for _, delta in events:
        count += delta
        max_count = max(max_count, count)
    
    return max_count
 
 
# ==================================================
# Example 2: Set State - Track Active Interval IDs
# ==================================================
 
def intervals_at_each_point(
    intervals: list[tuple[int, int]], 
    points: list[int]
) -> dict[int, list[int]]:
    """
    For each query point, return LIST of interval indices containing it.
    State: set of active interval IDs
    """
    events = []
    
    for i, (start, end) in enumerate(intervals):
        events.append((start, 0, 'START', i))  # 0 = process first
        events.append((end, 2, 'END', i))       # 2 = process last
    
    for j, p in enumerate(points):
        events.append((p, 1, 'QUERY', j))       # 1 = process middle
    
    events.sort()
    
    active = set()  # State: set of active interval IDs
    results = {}
    
    for pos, _, event_type, idx in events:
        if event_type == 'START':
            active.add(idx)
        elif event_type == 'END':
            active.discard(idx)
        elif event_type == 'QUERY':
            results[points[idx]] = list(active)
    
    return results
 
 
# ==================================================
# Example 3: Ordered State - Skyline Problem
# ==================================================
 
def get_skyline(buildings: list[list[int]]) -> list[list[int]]:
    """
    The Skyline Problem: given buildings as [left, right, height],
    return the skyline silhouette as list of [x, height] key points.
    
    State: multiset of active building heights (ordered)
    At each x, the skyline height = max of active heights (or 0 if none)
    
    Uses SortedList for O(log n) insertion/removal of heights.
    """
    # Events: (x, type, height)
    # type: 0 = building starts (entering from left)
    #       1 = building ends (exiting to right)
    events = []
    
    for left, right, height in buildings:
        events.append((left, 0, height))   # Start: add height
        events.append((right, 1, height))  # End: remove height
    
    # Sort: by x, then starts before ends, then taller starts first
    events.sort(key=lambda e: (e[0], e[1], -e[2] if e[1] == 0 else e[2]))
    
    # State: sorted list of active heights
    # We use a sorted list to easily get max
    active_heights = SortedList([0])  # 0 as sentinel for ground level
    
    result = []
    prev_max_height = 0
    
    for x, event_type, height in events:
        if event_type == 0:  # Building starts
            active_heights.add(height)
        else:  # Building ends
            active_heights.remove(height)
        
        # Current max height
        current_max = active_heights[-1]
        
        # If max height changed, record key point
        if current_max != prev_max_height:
            result.append([x, current_max])
            prev_max_height = current_max
    
    return result
 
 
# ==================================================
# Example 4: Counter per Y-coordinate (for area computation)
# ==================================================
 
def rectangles_area_union(rectangles: list[list[int]]) -> int:
    """
    Compute total area of union of axis-aligned rectangles.
    
    Sweep along x-axis, maintaining active y-ranges.
    At each x-event, compute change in covered area.
    
    Simplified version using coordinate compression.
    """
    # Collect all y-coordinates for compression
    y_coords = set()
    events = []
    
    for x1, y1, x2, y2 in rectangles:
        y_coords.add(y1)
        y_coords.add(y2)
        events.append((x1, 0, y1, y2))  # LEFT edge (add range)
        events.append((x2, 1, y1, y2))  # RIGHT edge (remove range)
    
    # Coordinate compression
    y_list = sorted(y_coords)
    y_to_idx = {y: i for i, y in enumerate(y_list)}
    
    # Count array: count[i] = number of active rectangles covering [y_list[i], y_list[i+1])
    count = [0] * len(y_list)
    
    events.sort()
    
    total_area = 0
    prev_x = events[0][0] if events else 0
    
    for x, event_type, y1, y2 in events:
        # Compute covered y-length
        covered_y = 0
        for i in range(len(y_list) - 1):
            if count[i] > 0:
                covered_y += y_list[i + 1] - y_list[i]
        
        # Add area since last x
        total_area += covered_y * (x - prev_x)
        prev_x = x
        
        # Update counts
        idx1, idx2 = y_to_idx[y1], y_to_idx[y2]
        delta = 1 if event_type == 0 else -1
        for i in range(idx1, idx2):
            count[i] += delta
    
    return total_area
 
 
# Example usage
print("Max overlap:", max_overlap_counter([(1, 5), (2, 6), (4, 8)]))
 
intervals = [(0, 10), (3, 7), (5, 15)]
points = [1, 5, 8, 12]
print("Intervals at points:", intervals_at_each_point(intervals, points))
 
buildings = [[2, 9, 10], [3, 7, 15], [5, 12, 12], [15, 20, 10], [19, 24, 8]]
print("Skyline:", get_skyline(buildings))

Choosing the Right State Structure

Counter: when you only need aggregate (count, sum, max). Set: when you need to know which intervals are active. Ordered structure (SortedList, BST): when you need the max/min of active values, or predecessor/successor queries. Segment tree: when you need efficient range queries on the state.

The Skyline Problem: A Classic Sweep Application

The Skyline Problem is a celebrated application of sweep line that appears in coding interviews and demonstrates the paradigm's power. Problem Statement: Given a list of buildings where each building is represented as [left, right, height] : left = x-coordinate of left edge right = x-coordinate of right edge height = building height Return the skyline formed by these buildings as a list of "key points" [x, height] where the height changes. The Sweep Line Approach: Events: Each building creates two events: (left, START, height) — building enters at x = left (right, END, height) — building exits at x = right State: A multiset (sorted list) of active building heights Processing: At each x-position, after updating the state, check if the maximum height changed. If so, record a key point. The Critical Insight: The skyline height at any x equals the maximum height among all buildings covering that x. By maintaining active heights in a max-queryable structure, we efficiently track this as the sweep progresses.

Skyline Problem Example

Computing the city skyline

Input

Buildings: [[2, 9, 10], [3, 7, 15], [5, 12, 12], [15, 20, 10], [19, 24, 8]]

Output

Skyline: [[2, 10], [3, 15], [7, 12], [12, 0], [15, 10], [20, 8], [24, 0]]

Explanation

At x=2, we rise to height 10. At x=3, building of height 15 starts (higher). At x=7, that building ends, dropping to 12. At x=12, last tall building ends, dropping to 0. And so on.

skyline.py
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
import heapq
from collections import defaultdict
 
def get_skyline(buildings: list[list[int]]) -> list[list[int]]:
    """
    Compute the skyline from a list of buildings.
    
    Uses a max-heap to track active building heights.
    Lazy deletion handles building exits.
    
    Time: O(n log n)
    Space: O(n)
    """
    if not buildings:
        return []
    
    # Events: (x, type, height)
    # type: 0 = start (entering), 1 = end (leaving)
    events = []
    
    for left, right, height in buildings:
        events.append((left, 0, height))   # Building starts
        events.append((right, 1, height))  # Building ends
    
    # Sort: by x, then starts before ends, then taller starts first
    # For ends: process shorter heights first (doesn't matter if using lazy deletion)
    events.sort(key=lambda e: (e[0], e[1], -e[2]))
    
    # Max-heap of active heights (use negative for max-heap with heapq)
    # We'll use lazy deletion: track counts of each height
    heap = [0]  # Start with ground level
    height_count = defaultdict(int)  # height -> active count
    height_count[0] = 1  # Ground always active
    
    result = []
    prev_max = 0
    
    i = 0
    while i < len(events):
        curr_x = events[i][0]
        
        # Process all events at the same x
        while i < len(events) and events[i][0] == curr_x:
            x, event_type, height = events[i]
            
            if event_type == 0:  # Start
                heapq.heappush(heap, -height)
                height_count[height] += 1
            else:  # End
                height_count[height] -= 1
            
            i += 1
        
        # Lazy deletion: pop heights that are no longer active
        while heap and height_count[-heap[0]] == 0:
            heapq.heappop(heap)
        
        # Current max height
        curr_max = -heap[0] if heap else 0
        
        # If max changed, record key point
        if curr_max != prev_max:
            result.append([curr_x, curr_max])
            prev_max = curr_max
    
    return result
 
 
# Cleaner version using SortedList (from sortedcontainers)
def get_skyline_sortedlist(buildings: list[list[int]]) -> list[list[int]]:
    """
    Skyline using SortedList for O(log n) max queries.
    Cleaner but requires external library.
    """
    from sortedcontainers import SortedList
    
    events = []
    for left, right, height in buildings:
        events.append((left, -height, 0))   # Start: negative height for sorting
        events.append((right, height, 1))   # End: positive height
    
    # Sort by x, then by value (starts before ends at same x, taller starts first)
    events.sort()
    
    active = SortedList([0])  # Active heights
    result = []
    prev_max = 0
    
    for x, h, event_type in events:
        if event_type == 0:  # Start
            active.add(-h)
        else:  # End
            active.remove(h)
        
        curr_max = active[-1]  # Max is last element in sorted list
        
        if curr_max != prev_max:
            result.append([x, curr_max])
            prev_max = curr_max
    
    return result
 
 
# Test
buildings = [[2, 9, 10], [3, 7, 15], [5, 12, 12], [15, 20, 10], [19, 24, 8]]
print("Skyline:", get_skyline(buildings))

Beyond Intervals: Geometric Sweep

Sweep line extends far beyond 1D intervals into 2D geometry and beyond. Line Segment Intersection (Bentley-Ottmann) Given a set of line segments, find all intersection points. Naive O(n²): Check every pair. Sweep O((n + k) log n): Where k = number of intersections Events: segment start, segment end, intersection (detected dynamically) State: balanced BST of active segments ordered by y-coordinate at current sweep x Key insight: only adjacent segments in the BST can intersect; swap them when they cross Closest Pair of Points Find the two closest points in a 2D plane. Sweep approach: Sort points by x-coordinate Maintain active points within distance d of current x (in a y-ordered structure) Only check active points for closer pairs Prune points that fall too far behind Voronoi Diagrams (Fortune's Algorithm) Construct Voronoi diagram using a sweep line, maintaining a "beach line" of parabolic arcs.

Advanced Sweep Line Applications

•Rectangle intersection detection — Find overlapping rectangle pairs in O((n + k) log n)
•Area of rectangle union — Covered area by overlapping rectangles
•Polygon triangulation — Decompose simple polygon into triangles
•Map overlay — Compute intersection of two planar subdivisions
•Visibility computation — What's visible from a point through obstacles
•Motion planning — Robot path planning using configuration space sweeps

The Sweep Principle Generalizes

Any problem where ordering by one dimension lets you process the remaining dimensions locally is a candidate for sweep line. The key question: 'Can I process objects incrementally as I move through space/time, updating a manageable state structure?'

Implementation Patterns and Best Practices

Let's codify the patterns that make sweep line implementations robust: Pattern 1: Event Class with Natural Ordering @dataclass(order=True) class Event: position: float priority: int # Tie-breaking data: Any = field(compare=False) # Not for ordering Pattern 2: Batch Processing at Same Position When multiple events share a position, process them all before updating the result: i = 0 while i < len(events): curr_pos = events[i].position while i < len(events) and events[i].position == curr_pos: process(events[i]) i += 1 record_result_at(curr_pos) Pattern 3: Lazy Deletion with Heaps When using heaps that don't support arbitrary removal: while heap and is_deleted(heap_top()): heapq.heappop(heap) current_value = heap_top() Pattern 4: Coordinate Compression When coordinates are sparse in a large range: coords = sorted(set(all_coordinates)) coord_to_index = {c: i for i, c in enumerate(coords)} # Now use indices instead of raw coordinates

sweep_patterns.py
Python
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
from dataclasses import dataclass, field
from typing import Any, Callable
from enum import IntEnum
import heapq
 
# =========================================
# Pattern: Generic Sweep Line Framework
# =========================================
 
class EventPriority(IntEnum):
    """Standard event priorities."""
    END = 0
    QUERY = 1  
    START = 2
 
 
@dataclass(order=True)
class SweepEvent:
    """Generic sweep event with natural ordering."""
    position: float
    priority: int  # Lower = processed first at same position
    data: Any = field(compare=False)
 
 
def sweep_line_generic(
    events: list[SweepEvent],
    on_start: Callable[[Any], None],
    on_end: Callable[[Any], None],
    on_query: Callable[[Any], Any],
) -> list:
    """
    Generic sweep line processor.
    
    Separates event generation from processing logic.
    """
    events.sort()
    results = []
    
    for event in events:
        if event.priority == EventPriority.START:
            on_start(event.data)
        elif event.priority == EventPriority.END:
            on_end(event.data)
        elif event.priority == EventPriority.QUERY:
            result = on_query(event.data)
            results.append(result)
    
    return results
 
 
# =========================================
# Example: Using the Framework
# =========================================
 
def count_intervals_at_points(intervals, query_points):
    """
    Use generic framework to count intervals at query points.
    """
    events = []
    
    for i, (start, end) in enumerate(intervals):
        events.append(SweepEvent(start, EventPriority.START, {'id': i}))
        events.append(SweepEvent(end, EventPriority.END, {'id': i}))
    
    for j, p in enumerate(query_points):
        events.append(SweepEvent(p, EventPriority.QUERY, {'point': p, 'idx': j}))
    
    # State
    count = 0
    results = [0] * len(query_points)
    
    def on_start(data):
        nonlocal count
        count += 1
    
    def on_end(data):
        nonlocal count
        count -= 1
    
    def on_query(data):
        results[data['idx']] = count
        return (data['point'], count)
    
    sweep_line_generic(events, on_start, on_end, on_query)
    return results
 
 
# =========================================
# Pattern: Heap with Lazy Deletion
# =========================================
 
class LazyMaxHeap:
    """
    Max-heap supporting lazy deletion.
    
    Items can be 'removed' without actually popping;
    they're skipped when accessed.
    """
    def __init__(self):
        self.heap = []  # Min-heap of negative values
        self.removed = {}  # value -> removal count
    
    def push(self, value):
        heapq.heappush(self.heap, -value)
    
    def remove(self, value):
        """Mark value for removal (lazy)."""
        self.removed[value] = self.removed.get(value, 0) + 1
    
    def _clean_top(self):
        """Remove lazily deleted items from top."""
        while self.heap:
            top = -self.heap[0]
            if self.removed.get(top, 0) > 0:
                heapq.heappop(self.heap)
                self.removed[top] -= 1
            else:
                break
    
    def max(self):
        """Return current maximum."""
        self._clean_top()
        return -self.heap[0] if self.heap else 0
    
    def pop(self):
        """Remove and return maximum."""
        self._clean_top()
        return -heapq.heappop(self.heap) if self.heap else 0
 
 
# Example: Skyline with lazy heap
def skyline_with_lazy_heap(buildings):
    """Skyline using LazyMaxHeap."""
    events = []
    for left, right, height in buildings:
        events.append((left, 0, height))
        events.append((right, 1, height))
    
    events.sort(key=lambda e: (e[0], e[1], -e[2] if e[1] == 0 else e[2]))
    
    heap = LazyMaxHeap()
    heap.push(0)  # Ground level
    
    result = []
    prev_max = 0
    
    i = 0
    while i < len(events):
        curr_x = events[i][0]
        
        while i < len(events) and events[i][0] == curr_x:
            x, event_type, height = events[i]
            if event_type == 0:
                heap.push(height)
            else:
                heap.remove(height)
            i += 1
        
        curr_max = heap.max()
        if curr_max != prev_max:
            result.append([curr_x, curr_max])
            prev_max = curr_max
    
    return result
 
 
# Test
print(count_intervals_at_points([(1, 5), (2, 6), (4, 8)], [0, 3, 5, 7]))
print(skyline_with_lazy_heap([[2, 9, 10], [3, 7, 15], [5, 12, 12]]))

Common Mistakes and Debugging

Sweep line bugs are often subtle. Here are the most common issues: Mistake 1: Wrong Event Priority Your algorithm gives wrong results when events coincide. Debug by: Listing tied events and the expected processing order Checking that your sort key produces this order Mistake 2: Off-By-One in State Update Updating state after recording instead of before, or vice versa. The general rule: Update state for the current event THEN query/record the result Mistake 3: Forgetting Ground Level / Base Case In skyline, forgetting to initialize with height 0 (ground level). The heap should never be truly empty. Mistake 4: Floating Point Comparison When positions are floats, equality checks need tolerance: if abs(event.position - current_position) < 1e-9: Mistake 5: Not Handling Empty State What's the max height when no buildings are active? The minimum overlap when no intervals? Always define the "nothing active" case.

Debugging Checklist

•Print events after sorting — Verify order is correct for tied positions
•Trace state at key positions — Add logging to show state changes
•Test with trivial cases — 0, 1, 2 intervals; no overlap; all overlap
•Test with ties — Events at same position reveal priority bugs
•Draw it out — Sketch intervals on a number line, manually trace the sweep
•Compare with brute force — For small inputs, verify against O(n²) solution

debugging_helper.py
Python
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
def debug_sweep(events, description="Sweep"):
    """
    Print events in processing order for debugging.
    """
    print(f"
=== {description} ===")
    print("Sorted events:")
    for e in sorted(events, key=lambda x: (x[0], x[1])):
        print(f"  Position {e[0]:6.2f}, Priority {e[1]}, Data: {e[2:]}")
    print()
 
 
def verify_with_brute_force(intervals, query_points, sweep_result):
    """
    Verify sweep result against naive O(n*m) computation.
    """
    brute_result = []
    for p in query_points:
        count = sum(1 for s, e in intervals if s <= p < e)
        brute_result.append(count)
    
    if sweep_result == brute_result:
        print("✓ Sweep result matches brute force")
    else:
        print("✗ Mismatch!")
        print(f"  Sweep:  {sweep_result}")
        print(f"  Brute:  {brute_result}")
        for i, p in enumerate(query_points):
            if sweep_result[i] != brute_result[i]:
                print(f"  Point {p}: Sweep={sweep_result[i]}, Brute={brute_result[i]}")
 
 
# Example usage
intervals = [(1, 5), (2, 6), (4, 8)]
points = [0, 1, 2, 3, 4, 5, 6, 7, 8]
 
# Create events for debugging
events = []
for i, (s, e) in enumerate(intervals):
    events.append((s, 0, 'START', i))
    events.append((e, 2, 'END', i))
for j, p in enumerate(points):
    events.append((p, 1, 'QUERY', j))
 
debug_sweep(events, "Interval Count Sweep")
 
# The actual sweep (simplified)
events_sorted = sorted(events)
active = 0
results = [0] * len(points)
 
for pos, priority, event_type, idx in events_sorted:
    if event_type == 'START':
        active += 1
    elif event_type == 'END':
        active -= 1
    elif event_type == 'QUERY':
        results[idx] = active
 
print(f"Sweep results: {results}")
verify_with_brute_force(intervals, points, results)

Event-Based Sweeping

The Sweep Line Paradigm

The Sweep Line Framework

Event Types and Priority Ordering

State Structures for Sweep Line

The Skyline Problem: A Classic Sweep Application

Beyond Intervals: Geometric Sweep

Implementation Patterns and Best Practices

Common Mistakes and Debugging

There’s more to understand.

Event-Based Sweeping

The Sweep Line Paradigm

The Sweep Line Framework

Event Types and Priority Ordering

State Structures for Sweep Line

The Skyline Problem: A Classic Sweep Application

Beyond Intervals: Geometric Sweep

Implementation Patterns and Best Practices

Common Mistakes and Debugging

There’s more to understand.