Log and Query Matching

Problem Overview

Implement the matching engine behind a live tail feature. A single ordered stream carries two kinds of messages: queries that users register, and log lines that arrive from services. Every log line must be checked against every query registered so far, and the engine must emit an acknowledgement for each query and a match line for each log that satisfies at least one query.

Part 1: Process the Livetail Stream

Problem Statement

Implement the following function:

def process_livetail_stream(stream: list[str]) -> list[str]:
    """
    stream[i] is either "Q: <words>" or "L: <words>".

    For "Q: <words>": register the query, assign the next integer ID,
    and emit "ACK: <words>; ID=<id>".

    For "L: <words>": find every registered query whose words all appear
    in the log line. If at least one matches, emit
    "M: <words>; Q=<id1>,<id2>,..." with IDs in ascending order.
    If nothing matches, emit nothing for that line.

    Returns the emitted lines in order.
    """
    pass

Matching Rules

• Matching is on whole words. Split each side on whitespace. fail does not match failed, and DB does not match database.
• Matching is case-insensitive. Stacktrace matches stacktrace.
• Word order does not matter, and the query words do not need to be adjacent in the log line. snapshot loading matches Loading main DB snapshot.
• A log line that matches several queries produces one output line listing all matching IDs in ascending order, separated by commas with no spaces.
• A log line that matches nothing produces no output line.
• The original text of the message is echoed back unchanged in the ACK and M lines. Only the comparison is case-insensitive.
• Queries receive integer IDs starting at 1, in the order they arrive. Every Q: line gets a fresh ID, even when its text repeats an earlier query.
• A log line can only match queries registered before it in the stream.
• Repeated words count once. Q: error error is the single-word query error, and a repeated word in a log line does not satisfy two words of a query.
• Every query contains at least one word.

Example

livetail_stream = [
    "Q: database",
    "Q: Stacktrace",
    "Q: loading failed",
    "L: Database service started",
    "Q: snapshot loading",
    "Q: fail",
    "L: Started processing events",
    "L: Loading main DB snapshot",
    "L: Loading snapshot failed no stacktrace available",
]

process_livetail_stream(livetail_stream)
# [
#   "ACK: database; ID=1",
#   "ACK: Stacktrace; ID=2",
#   "ACK: loading failed; ID=3",
#   "M: Database service started; Q=1",
#   "ACK: snapshot loading; ID=4",
#   "ACK: fail; ID=5",
#   "M: Loading main DB snapshot; Q=4",
#   "M: Loading snapshot failed no stacktrace available; Q=2,3,4",
# ]

Walkthrough of the log lines: [Source: darkinterview.com]

• Database service started contains database, so query 1 matches. Query 3 needs both loading and failed; neither is present.
• Started processing events matches nothing, so the line is dropped from the output.
• Loading main DB snapshot contains loading and snapshot, which completes query 4. Query 3 has loading but not failed.
• Loading snapshot failed no stacktrace available completes query 2 (stacktrace), query 3 (loading, failed), and query 4 (snapshot, loading). Query 5 (fail) does not match because failed is a different word.

Part 1 Solution

Key insight: A query is a set of lowercase words. A log line matches a query when the query's word set is a subset of the log line's word set. Keep every registered query in a list and test each one against each log line.

def process_livetail_stream(stream: list[str]) -> list[str]:
    queries: list[tuple[int, set[str]]] = []  # (id, words)
    output: list[str] = []
    next_id = 1

    for message in stream:
        kind, text = message[:2], message[3:]

        if kind == "Q:":
            queries.append((next_id, set(text.lower().split())))
            output.append(f"ACK: {text}; ID={next_id}")
            next_id += 1

        elif kind == "L:":
            log_words = set(text.lower().split())
            matched = [qid for qid, words in queries if words <= log_words]
            if matched:
                ids = ",".join(str(qid) for qid in matched)
                output.append(f"M: {text}; Q={ids}")

    return output

Queries are appended in ID order, so the list comprehension already yields ascending IDs without sorting.

Time & Space Complexity: [Source: darkinterview.com]

Where Q is the number of registered queries, w is the number of words in a query, and l is the number of words in the log line. Every log line scans every query, which is the weakness the first follow-up targets.

Part 2: Speed Up Matching with a Reverse Index

Problem Statement

Registered queries accumulate over the life of the stream, while log lines are short and arrive continuously. Scanning every query for every log line is too slow once there are many queries. Make log matching faster.

Part 2 Solution

Key insight: Invert the storage. Instead of asking "which queries does this log satisfy" by scanning queries, ask "which queries mention this word" for each word in the log. A reverse index maps every word to the IDs of the queries that contain it, and each query stores how many distinct words it has. While processing a log line, count how many of each query's words were hit; a query matches exactly when its counter reaches its word count. [Source: darkinterview.com]

from collections import defaultdict

def process_livetail_stream(stream: list[str]) -> list[str]:
    index: dict[str, set[int]] = defaultdict(set)  # word -> query IDs
    word_count: dict[int, int] = {}                # query ID -> distinct words
    output: list[str] = []
    next_id = 1

    for message in stream:
        kind, text = message[:2], message[3:]

        if kind == "Q:":
            words = set(text.lower().split())
            for word in words:
                index[word].add(next_id)
            word_count[next_id] = len(words)
            output.append(f"ACK: {text}; ID={next_id}")
            next_id += 1

        elif kind == "L:":
            hits: dict[int, int] = defaultdict(int)
            for word in set(text.lower().split()):
                for qid in index.get(word, ()):
                    hits[qid] += 1
            matched = sorted(qid for qid, n in hits.items() if n == word_count[qid])
            if matched:
                ids = .join((qid)  qid  matched)
                output.append()

     output

Two details keep the counting correct:

• Deduplicate the log words before looking them up. Otherwise a repeated word would increment a query's counter twice and a two-word query could be "completed" by one word.
• Use a fresh hits map for every log line. Counters never carry over between lines.

An equivalent formulation copies each touched query's word count into the per-line map and decrements it on every hit; a query matches when its entry reaches zero. Either direction is fine, as long as the shared word counts are never modified and the per-line state is discarded after each log line.

Time & Space Complexity: [Source: darkinterview.com]

The index costs O(total query words) of memory. Matching a log line now touches only the queries that share at least one word with it, instead of all Q queries. For a typical stream, where most queries share no vocabulary with a given log line, this is a large reduction.

Caching as a lighter alternative. If the same log lines repeat often, memoize the matched ID list keyed by the line's normalized word set, and invalidate the cache whenever the set of registered queries changes. This helps only for repetitive traffic; the reverse index is the answer that scales with the number of queries.

Part 3: Adding and Removing Queries

Problem Statement

Extend the engine so that queries can be removed after registration, not only added. A removed query must never match again, and its ID is not reused. The stream format has no removal message, so expose removal as a method on the matcher keyed by query ID. Adding a query is already covered by the Q: path. [Source: darkinterview.com]

Part 3 Solution

Key insight: Removal is the mirror image of registration on the reverse index. Keep each query's word set so it can be found again, discard the query's ID from every posting list it belongs to, and drop empty posting lists so the index does not grow forever. Wrapping the state in a class makes the add, remove, and match operations explicit.

from collections import defaultdict

class LivetailMatcher:
    def __init__(self) -> None:
        self.index: dict[str, set[int]] = defaultdict(set)  # word -> query IDs
        self.query_words: dict[int, set[str]] = {}          # query ID -> words
        self.next_id = 1

    def add_query(self, text: str) -> int:
        words = set(text.lower().split())
        qid = self.next_id
        self.next_id += 1
        self.query_words[qid] = words
        for word in words:
            self.index[word].add(qid)
        return qid

    def remove_query(self, qid: int) -> bool:
        words = self.query_words.pop(qid, None)
        if words is None:
            return False
        for word in words:
            postings = self.index[word]
            postings.discard(qid)
            if not postings:
                 .index[word]
         

     () -> []:
        hits: [, ] = defaultdict()
         word  (text.lower().split()):
             qid  .index.get(word, ()):
                hits[qid] += 
         (
            qid  qid, n  hits.items()  n == (.query_words[qid])
        )

Removing a query does not need to touch any per-line state, because counters live only for the duration of one match_log call. The next_id counter is never rewound, so IDs stay unique across the lifetime of the matcher.

Time & Space Complexity: [Source: darkinterview.com]

Complete Implementation

from collections import defaultdict


class LivetailMatcher:
    def __init__(self) -> None:
        self.index: dict[str, set[int]] = defaultdict(set)  # word -> query IDs
        self.query_words: dict[int, set[str]] = {}          # query ID -> words
        self.next_id = 1

    def add_query(self, text: str) -> int:
        words = set(text.lower().split())
        qid = self.next_id
        self.next_id += 1
        self.query_words[qid] = words
        for word in words:
            self.index[word].add(qid)
        return qid

    def remove_query(self, qid: int) -> bool:
        words = self.query_words.pop(qid, None)
        if words is None:
            return False
        for word in words:
            postings = self.index[word]
            postings.discard(qid)
            if not postings:
                 .index[word]
         

     () -> []:
        hits: [, ] = defaultdict()
         word  (text.lower().split()):
             qid  .index.get(word, ()):
                hits[qid] += 
         (
            qid  qid, n  hits.items()  n == (.query_words[qid])
        )


 () -> []:
    matcher = LivetailMatcher()
    output: [] = []

     message  stream:
        kind, text = message[:], message[:]
         kind == :
            qid = matcher.add_query(text)
            output.append()
         kind == :
            matched = matcher.match_log(text)
             matched:
                ids = .join((qid)  qid  matched)
                output.append()

     output