Group duplicate records that share any property

Treat every record as a node in a graph. Two records are joined by an edge
when they hold the same value in the same field. A group is a connected
component of that graph. The relation is transitive through chains: if A
shares a field with B and B shares a different field with C, then A, B and C
form one group even though A and C have nothing in common.

Brute force. Compare every pair of records and add an edge when any field
matches, then run BFS or DFS over the edges. That is O(n²) comparisons and up
to O(n²) edges. It is correct but too slow for a large list, and it is the
baseline to state before you improve it.

Optimal: union-find keyed by field value. You do not need every pair. For
each field, keep a map from value to the first record that held it. When a
later record holds the same value, union it with that first record. Every
record with a given value then joins one representative, so the value's
whole set is connected with k - 1 unions instead of k² comparisons. Edges
between records via different values chain together through the union-find
structure.

Key the map by (field_index, value), not by value alone. A name of
"x" and an email of "x" are not the same property. Skip empty values
(None or ""), because a missing field must not join records.

Invariants:

• After record i is processed, any two records among the first i+1 that are
  linked through a chain of shared field values have the same root.
• first_owner[(f, v)] always names a record that is already in the
  component of every record seen so far with value v in field f.

Union by size and path compression keep each operation at amortized
α(n), which is effectively constant.

At the end, bucket record IDs by root. Return each group sorted, and order the
groups by their smallest ID so the output is deterministic.

from collections import defaultdict
from typing import Hashable, Optional, Sequence


class UnionFind:
    def __init__(self, n: int) -> None:
        self.parent = list(range(n))
        self.size = [1] * n

    def find(self, x: int) -> int:
        root = x
        while self.parent[root] != root:
            root = self.parent[root]
        while self.parent[x] != root:  # path compression
            self.parent[x], x = root, self.parent[x]
        return root

    def union(self, a: int, b: int) -> None:
        ra, rb = self.find(a), self.find(b)
        if ra == rb:
            return
        if self.size[ra] < self.size[rb]:
            ra, rb = rb, ra
        self.parent[rb] = ra
        self.size[ra] += self.size[rb]


Record = tuple[int, Optional[Hashable], Optional[Hashable], Optional[Hashable]]


def group_records(records: Sequence[Record]) -> list[list[int]]:
    """Group record IDs. Records share a group when they are linked by a
    chain of records that hold the same value in the same field."""
    n = len(records)
    uf = UnionFind(n)
    first_owner: dict[tuple[int, Hashable], int] = {}

    for i, (_rid, *props) in enumerate(records):
        for field, value in enumerate(props):
            if value is None or value == "":
                continue
            key = (field, value)
            owner = first_owner.get(key)
            if owner is None:
                first_owner[key] = i
            else:
                uf.union(owner, i)

    buckets: dict[int, list[int]] = defaultdict(list)
    for i, (rid, *_props) in enumerate(records):
        buckets[uf.find(i)].append(rid)

    groups = [sorted(ids) for ids in buckets.values()]
    groups.sort(key=lambda g: g[0])
    return groups


if __name__ == "__main__":
    recs: list[Record] = [
        (1, "ana", "ana@x.io", "555-01"),
        (2, "ben", "ben@x.io", "555-02"),
        (3, "cy", "ana@x.io", "555-03"),
        (4, "dee", "dee@x.io", "555-02"),
        (5, "eli", "eli@x.io", "555-05"),
    ]
    assert group_records(recs) == [[1, 3], [2, 4], [5]]

    chain: list[Record] = [
        (10, "a", "m1", "p1"),
        (11, "b", "m1", "p2"),
        (12, "c", "m2", "p2"),
        (13, "c", "m3", "p3"),
    ]
    assert group_records(chain) == [[10, 11, 12, 13]]

    assert group_records([]) == []
    assert group_records([(7, "x", "y", "z")]) == [[7]]
    # Same value in different fields does not match.
    assert group_records([(1, "x", "a", "b"), (2, "c", "x", "d")]) == [[1], [2]]
    # Missing values never join records.
    assert group_records([(1, None, "", "p"), (2, None, "", "q")]) == [[1], [2]]
    print("ok")

• Let n be the number of records and k the number of fields (3 here).
• Building the groups: O(n·k·α(n)) time, which is effectively O(n) for fixed
  k, plus O(n log n) to sort the output deterministically. Drop the sort if
  any order is acceptable.
• Space: O(n·k) for the value map, O(n) for the union-find arrays and the
  buckets.
• Brute force for comparison: O(n²·k) time.

• [(1,"ana","ana@x.io","555-01"), (2,"ben","ben@x.io","555-02"), (3,"cy","ana@x.io","555-03"), (4,"dee","dee@x.io","555-02"), (5,"eli","eli@x.io","555-05")]
  → [[1, 3], [2, 4], [5]] (1 and 3 share an email, 2 and 4 share a phone).
• Transitive chain [(10,"a","m1","p1"), (11,"b","m1","p2"), (12,"c","m2","p2"), (13,"c","m3","p3")]
  → [[10, 11, 12, 13]]; 10 and 13 share nothing directly.
• Empty input [] → [].
• One record [(7,"x","y","z")] → [[7]].
• Same value in different fields [(1,"x","a","b"), (2,"c","x","d")]
  → [[1], [2]].
• Missing values [(1,None,"","p"), (2,None,"","q")] → [[1], [2]].
• All records share field 1 → one group that holds every ID.
• No shared values anywhere → n singleton groups.

• Records arrive as a stream; answer "how many groups now?" after each
  one. Keep the union-find and the value map alive between calls, and keep
  a component counter that starts at +1 per new record and drops by 1 on each
  successful union. Each insert is amortized O(k·α(n)).
• Return only the groups with more than one record (the actual
  duplicates), or one canonical record per group. Filter buckets by size,
  or pick a representative by a rule (earliest ID, most complete record).
  Track that representative at the root during union so no second pass is
  needed.
• A value matches across fields (for example, a phone number stored in the
  wrong column), or there are m fields instead of 3. Key the map by the
  normalized value only, and loop over any number of fields. Normalize first
  (lowercase emails, strip punctuation from phones), because matching on raw
  strings misses obvious duplicates.
• Records can be deleted. Union-find does not support splits. Either
  rebuild lazily when a delete happens, or keep per-value record sets and
  recompute only the affected component with BFS over the value index. Say
  which you choose based on the delete rate.
• The data is too large for one machine. Run connected components as a
  distributed job: emit (field, value) → record pairs, group by key to get
  edges to a representative, then run iterative label propagation (or
  a union-find per shard, then merge the shard boundaries) until labels stop
  changing.

Mid-level (L4)

• Sees that matches chain transitively, so this is connected components and
  not pairwise duplicate detection.
• Starts from the O(n²) pairwise approach, then reaches union-find or a
  value-to-records index with BFS.
• Writes working code that handles empty input and singletons.

Senior (L5)

• Avoids the O(n²) edge set by unioning each record with the first owner of
  each value, and explains why that keeps the component structure the same.
• Keys by (field, value) and handles missing values without being
  prompted. Implements path compression and union by size.
• States the complexity precisely and handles the streaming count follow-up
  without rewriting the solution.

Staff (L6+)

• Asks about the data first: normalization, false merges from shared values
  such as a common office phone, and whether one bad value should be allowed
  to merge two large groups.
• Discusses deletions, the limits of union-find for them, and the trade-offs
  of rebuilding.
• Outlines a distributed connected-components design and how to make it
  converge and stay cheap on skewed values that appear in very many records.