Answer Policy Violation Log Queries
PinterestSoftware Engineer · Technical ScreenLast updated
Design an in-memory query engine over a timestamp-sorted policy violation log to support aggregate, point, and range queries efficiently using hash maps and binary search.
Problem Statement
You are tasked with building a policy violation tracking system for content moderation. The system is initialized with an append-only log of moderation violations, where each log record contains [post_id, policy, timestamp]. The records are guaranteed to be sorted in non-decreasing order of timestamp.
Implement the PolicyViolationTracker class with the following methods:
-
__init__(logs: list[list[str, str, int]]): Initializes the tracker with the given violation logs. Each record is[post_id, policy, timestamp]. Multiple violations can share the same timestamp, and a single post may violate multiple policies (or the same policy multiple times). -
count_unique_posts_by_policy(policy: str) -> int: Returns the total number of distinct posts that have violated the specifiedpolicy. -
get_policies_for_post(post_id: str) -> list[str]: Returns a list of all distinct policies that the specifiedpost_idhas violated, sorted in lexicographical order. Returns an empty list if the post has no violations. -
count_unique_posts_at_time(timestamp: int) -> int: Returns the count of distinct posts that incurred at least one violation at the exacttimestamp. -
count_unique_posts_in_range(start_time: int, end_time: int) -> int: Returns the count of distinct posts that incurred at least one violation within the inclusive range[start_time, end_time]. -
count_violations_by_policy_in_range(policy: str, start_time: int, end_time: int) -> int: Returns the total number of violation records (including repeated violations by the same post) for the givenpolicywithin the inclusive range[start_time, end_time].
Example
logs = [
["post_1", "spam", 100],
["post_2", "copyright", 100],
["post_1", "hate_speech", 150],
["post_3", "spam", 200],
["post_1", "spam", 250],
]
tracker = PolicyViolationTracker(logs)
tracker.count_unique_posts_by_policy("spam")
# Output: 2 (post_1 and post_3)
tracker.get_policies_for_post("post_1")
# Output: ["hate_speech", "spam"] (distinct, lexicographical)
tracker.count_unique_posts_at_time(100)
# Output: 2 (post_1 and post_2)
tracker.count_unique_posts_in_range(100, 150)
# Output: 2 (post_1 and post_2)
tracker.count_violations_by_policy_in_range("spam", 100, 250)
# Output: 3 (post_1 at 100, post_3 at 200, post_1 at 250)
Constraints
1 <= len(logs) <= 10^5post_idandpolicyare non-empty strings of alphanumeric characters and underscores with length<= 64.0 <= timestamp <= 10^14logsis sorted in non-decreasing order bytimestamp.start_time <= end_timefor all range queries.- Up to
2 * 10^4total calls will be made across all query methods.
Hints
Hint 1
Pre-aggregating sets during initialization allows the point lookups (by policy, post, or exact timestamp) to be answered in O(1) or O(k log k) time where k is the number of distinct policies.
Hint 2
For `count_violations_by_policy_in_range`, keep a sorted list of timestamps for each policy and apply binary search (bisect) to find the range boundaries in O(log N).
Hint 3
For `count_unique_posts_in_range`, since the logs are already sorted by timestamp, binary search can identify the slice of logs in [start_time, end_time], after which you can collect unique post IDs.
Follow-up questions
- How would you optimize `count_unique_posts_in_range` if range queries are extremely frequent and the time window spans thousands of records?
- If logs arrive continuously in real-time instead of being pre-loaded, how would your data structures and thread-safety design change?
- How would you scale this system if the logs do not fit in the memory of a single machine?
Solution walkthrough is for members
Get the full walkthrough with reference code and complexity analysis, plus Kitty Coach at 300 messages a day. Hints, the code runner and community answers stay free.