Katkat

Answer Policy Violation Log Queries

PinterestSoftware Engineer · Technical Screen

Last 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 specified policy.

  • get_policies_for_post(post_id: str) -> list[str]: Returns a list of all distinct policies that the specified post_id has 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 exact timestamp.

  • 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 given policy within 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^5
  • post_id and policy are non-empty strings of alphanumeric characters and underscores with length <= 64.
  • 0 <= timestamp <= 10^14
  • logs is sorted in non-decreasing order by timestamp.
  • start_time <= end_time for all range queries.
  • Up to 2 * 10^4 total 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?