Skip to main content

Data Structure Pairing: Deque + Hash Map (Rolling-Window Dedup)

πŸ”‘ Key Insight​

The hash map is always the source of truth β€” it stores the authoritative state (uniqueness, counts, last-seen timestamps). The deque is just a helper β€” it maintains arrival order so we know which elements to evict as they age out.

πŸ‘‰ This means we always check the hash map first to decide whether an event should be accepted. The deque is only used to know when to decrement or remove entries.


🧩 Problem​

How can we efficiently count unique IDs in a sliding time window (e.g., the last 10 seconds)?

NaΓ―ve approaches break down:

  • Queue only: Easy eviction by time, but slow uniqueness checks (for insertion)
  • Hash map only: O(1) lookups, but no fast eviction (requires linear scan).
  • Deque + Hash map: Combines both strengths β€” ordered eviction and constant-time lookup.

🍷 Alfredo Pairing: Deque + Hash Map​

This pairing is like wine with pasta β€” better together. One structure maintains order, the other provides fast membership. Used together, they solve rolling-window deduplication elegantly.

When to Use​

  • Rate limiting
  • Spam filtering
  • Session/user deduplication
  • Analytics counters over rolling time windows

βœ… Python Implementation (Uniqueness)​

from dataclasses import dataclass
from collections import deque
from typing import Deque
import time

@dataclass
class Event:
user_id: str
timestamp: float

class RollingUniqueCounter:
def __init__(self, window_seconds: int):
self.window = window_seconds
self.events: Deque[Event] = deque() # Tracks arrival order for eviction
self.active_users: dict[str, float] = {} # Tracks active IDs β†’ last_seen_timestamp

def insert(self, user_id: str):
now = time.time()
cutoff = now - self.window

# Evict expired
while self.events and self.events[0].timestamp < cutoff:
evict_event = self.events.popleft()
if self.active_users.get(evict_event.user_id) == evict_event.timestamp:
del self.active_users[evict_event.user_id]

# Update hashmap
self.active_users[user_id] = now

# Append obj to Deque
self.events.append(Event(user_id, now))

def count_unique(self) -> int:
return len(self.active_users)

⏱️ Complexity​

  • Insert: O(1) amortized
  • Evict: O(1) amortized
  • Count: O(1)

The combined structure guarantees constant-time operations while staying memory-bounded by the size of the window.


🚦 Rate Limiter Extension (Per-User Counts)​

A natural extension is a per-user rate limiter, e.g., β€œmax 3 requests per second per user.”

from dataclasses import dataclass
from collections import deque
from typing import Deque
import time

@dataclass
class Event:
user_id: str
timestamp: float

class RateLimiter:
def __init__(self, window_seconds: int, max_requests: int):
self.window = window_seconds
self.max_requests = max_requests
self.events: Deque[Event] = deque() # Tracks arrival order for eviction
self.user_hashmap = {} # user_id -> {count: int, last_seen: float}

def allow(self, user_id: str) -> bool:
now = time.time()
cutoff = now - self.window

# Evict expired
while self.events and self.events[0].timestamp < cutoff:
user_id, _ = self.events.popleft()
if user_id in self.user_hashmap:
self.user_hashmap[user_id]["count"] -= 1
if self.user_hashmap[user_id]["count"] == 0:
del self.user_hashmap[user_id]

# Check hash map first (source of truth)
if self.user_hashmap.get(user_id, {"count": 0})["count"] >= self.max_requests:
return False

# Accept request β†’ update state
self.events.append((user_id, now))
if user_id not in self.user_hashmap:
self.user_hashmap[user_id] = {"count": 0, "timestamps": []}
self.user_hashmap[user_id]["count"] += 1
self.user_hashmap[user_id]["timestamps"].append(now) # while we dont need to do this, it would allow for a ton of custom logic (with the tradeoff of consuming more memory)
return True