Katkat

Propagate Task Filtering Through Ancestors

RipplingSoftware Engineer · Technical Screen

Last updated

Design an algorithm to filter a hierarchy of tasks where any task failing a predicate—or having any ancestor that fails or is missing—is excluded along with all its descendants, running in linear time and preserving original order.

Propagate Task Filtering Through Ancestors

You are given a list of task records representing a hierarchy. Each task is represented as a dictionary with at least the following keys:

  • "id" (str): A unique identifier for the task.
  • "parent_id" (Optional[str]): The id of its direct parent task, or None if it is a root task.

You are also given a filtering function predicate(task: dict) -> bool. A task is considered directly valid if predicate(task) returns True.

However, tasks exist in a dependency hierarchy. A task must be retained in the final output if and only if:

  1. It is directly valid (predicate(task) == True).
  2. Every ancestor in its chain of parents is also retained.
  3. It does not belong to a cycle.
  4. Its parent_id, if not None, exists in the input list.

If any ancestor is invalid, missing, or part of a cycle, all of its descendants must be filtered out, even if they individually satisfy predicate.

Tasks in the input list may appear in any arbitrary order (children can appear before their parents).

Write a function filter_task_hierarchy(tasks: list[dict], predicate: Callable[[dict], bool]) -> list[dict] that returns only the retained tasks, preserving their original relative order in the input list.

Function Signature

from typing import Callable, Optional

def filter_task_hierarchy(
    tasks: list[dict],
    predicate: Callable[[dict], bool]
) -> list[dict]:
    pass

Constraints

  • 0 <= len(tasks) <= 100,000
  • Task id values are unique strings.
  • parent_id is either None or a string.
  • The algorithm must run in O(N) average time and O(N) auxiliary space.

Examples

Example 1

tasks = [
    {"id": "child", "parent_id": "parent", "name": "Subtask A"},
    {"id": "unrelated", "parent_id": None, "name": "Standalone Task"},
    {"id": "grandparent", "parent_id": None, "name": "Project Root"},
    {"id": "parent", "parent_id": "grandparent", "name": "Feature 1"},
]

# Filter out 'grandparent'
predicate = lambda t: t["id"] != "grandparent"

assert filter_task_hierarchy(tasks, predicate) == [
    {"id": "unrelated", "parent_id": None, "name": "Standalone Task"}
]

Explanation: While parent and child pass the predicate, their ancestor grandparent fails. Thus, parent and child are pruned. unrelated has no parent and passes the predicate, so it is kept.

Example 2

tasks = [
    {"id": "A", "parent_id": "B"},
    {"id": "B", "parent_id": "A"},
    {"id": "C", "parent_id": None},
]
predicate = lambda t: True

assert filter_task_hierarchy(tasks, predicate) == [
    {"id": "C", "parent_id": None}
]

Explanation: Tasks A and B form a cycle and have no valid root ancestor, so both are excluded.

Hints

Hint 1

Build a fast lookup map from task id to task record so parent lookups can be performed in O(1) time.

Hint 2

Use memoization (with states like UNVISITED, VISITING, VALID, INVALID) to evaluate whether a task is retained in O(1) amortized time while detecting cycles.

Hint 3

After computing whether each task is valid using memoized traversal, do a final linear scan over the original input array to collect tasks in their original order.

Follow-up questions

  • How would you handle a scenario where the hierarchy is distributed across partitions and cannot fit into a single machine's memory?
  • If updates to task statuses arrive in a streaming fashion, how would you design an incremental data structure to update descendant retention without recomputing the entire tree?
  • How would your approach change if tasks could have multiple parents (Directed Acyclic Graph) rather than a single parent tree?