Implement a Unique Lines Command
VantaSoftware Engineer · OnsiteLast updated
Implement a streaming generator that mimics the Unix `uniq` command by collapsing consecutive duplicate lines with options for counting and case-insensitivity.
Problem Statement
In Unix-like operating systems, the uniq utility reads an input stream of text lines and filters out adjacent duplicate lines.
Implement a generator function unique_lines that accepts an iterable of lines and yields the deduplicated stream according to the specified options.
Function Signature
from typing import Iterable, Iterator
def unique_lines(
stream: Iterable[str],
ignore_case: bool = False,
show_count: bool = False
) -> Iterator[str]:
pass
Requirements
- Consecutive Deduplication: Only adjacent duplicates should be collapsed. Non-adjacent duplicates should still appear.
- Case Sensitivity: If
ignore_caseisTrue, lines should be compared case-insensitively (e.g.,"Foo"and"foo"are considered duplicates). However, the output line should preserve the exact casing of the first occurrence in the consecutive run. - Counts: If
show_countisTrue, format the output string as"{count} {line}", wherecountis the number of consecutive occurrences of that line. - Streaming: The input stream may be arbitrarily large or infinite. Your implementation must process lines lazily with
O(1)auxiliary space beyond storing the current line and count.
Constraints
- Line lengths:
0 <= len(line) <= 10^5characters. - Number of lines in
stream:0 <= N <= 10^7(or unbounded stream). - Auxiliary space complexity must be
O(1)(excluding the yielded values).
Examples
Example 1
stream = ["apple", "apple", "banana", "apple"]
result = list(unique_lines(stream))
assert result == ["apple", "banana", "apple"]
Explanation: The first two "apple" lines are adjacent duplicates and collapse into one. The later "apple" is not adjacent to the previous ones, so it is preserved.
Example 2
stream = ["System", "system", "SYSTEM", "Error", "error"]
result = list(unique_lines(stream, ignore_case=True, show_count=True))
assert result == ["3 System", "2 Error"]
Explanation: "System", "system", and "SYSTEM" form a single run of 3 occurrences; the casing of the first line ("System") is retained. "Error" and "error" form a run of 2 occurrences.
Hints
Hint 1
Maintain state for the representative line of the current run, its normalized comparison key, and the run's occurrence count.
Hint 2
When encountering a line that differs from the current run's key, emit the formatted output for the previous run before resetting the state.
Hint 3
Ensure you emit the final run after the stream is fully consumed (or handle empty inputs gracefully without emitting anything).
Follow-up questions
- How would you modify the function to support flags like `-u` (only print lines that were never duplicated) and `-d` (only print lines that were duplicated)?
- If the input lines are not sorted and you need to remove global duplicates from a file larger than available RAM, how would you design an external merge-sort solution?
- How would you handle streaming data where individual lines themselves might be too large to fit in memory?
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.