Summary

Adds a local pHash-based deduplication utility for Nucleus SDK users who already have a large list of DatasetItem objects or rows from items_and_annotation_generator().

• Exposes nucleus.deduplicate_by_phash(...) and LocalDeduplicationResult.
• Accepts either DatasetItem objects or generator rows with an "item" DatasetItem.
• Preserves Nucleus threshold semantics for 0..64.
• Uses fast exact-pHash handling for threshold == 0, a single-item fast path for threshold == 64, an exact chunked Hamming-neighbor index for 1 <= threshold <= 10, and a linear exact fallback for thresholds above 10.
• Returns the surviving original objects, surviving dataset items, reference IDs, and DeduplicationStats.

Writeup: Local pHash Deduplication

Baseline Problem

The logical algorithm is greedy deduplication. Sort rows deterministically, then keep a candidate only if no previously kept pHash is within the requested Hamming distance threshold.

kept = []
for row in sorted_rows:
    if not any(hamming(row.phash, kept_hash) <= threshold for kept_hash in kept):
        kept.append(row)

This is exact but effectively quadratic for mostly distinct inputs, because each new candidate may scan every pHash kept so far.

Technique 1: Fail Fast and Normalize Once

Before the dedup pass starts, each input is normalized into an internal record with the original object, its DatasetItem, a stable tie-break ID, and the pHash parsed as a 64-bit integer. Threshold validation and pHash validation happen up front, so a long local run fails before doing expensive work if any pHash is missing or malformed.

Technique 2: Integer pHashes and Integer Hamming Distance

Backend pHashes arrive as 64-character binary strings. The utility parses each pHash once with int(phash, 2), then computes Hamming distance with:

(candidate_phash_value ^ kept_phash_value).bit_count()

This avoids per-character string comparison and enables sorting by (phash_value, stable_id).

Technique 3: Threshold-Specific Fast Paths

Technique 4: Chunking With the Pigeonhole Principle

The accelerated path splits each 64-bit pHash into six chunks:

INDEX_CHUNK_BITS = (11, 11, 11, 11, 10, 10)

For threshold t, the per-chunk radius is floor(t / 6). If two pHashes have total Hamming distance at most t, at least one of the six chunks must be within that per-chunk radius. Otherwise every chunk would exceed the radius, forcing the total Hamming distance above t.

For threshold=10, the per-chunk radius is 1. One partition performs 4 * 12 + 2 * 11 = 70 dictionary lookups: each 11-bit chunk checks the exact value plus 11 one-bit variants, and each 10-bit chunk checks the exact value plus 10 one-bit variants.

Technique 5: Two Partitions With Cheap Rotation

A single chunk partition can produce many false candidates. The utility uses two exact partitions:

• Partition 0 splits the original 64-bit pHash into six contiguous chunks.
• Partition 1 rotates the pHash right by 8 bits, then splits into the same chunk widths.

The rotation is cheap:

((phash_value >> 8) | (phash_value << 56)) & ((1 << 64) - 1)

The pigeonhole argument applies independently to both partitions, so true duplicates cannot be missed. A candidate must be found by both partitions before the exact 64-bit Hamming check runs.

Technique 6: Reusable Candidate Marks Instead of Set Intersections

The permanent state remains simple: kept_hashes stores pHashes that survived deduplication, and chunk dictionaries point chunk values back to positions in kept_hashes. Those positions are called kept indexes.

Conceptually, each query intersects two candidate sets:

partition_0_candidates = {17, 31, 44, 91, ...}
partition_1_candidates = {3, 17, 88, 91, ...}
intersection = {17, 91, ...}

The first implementation literally allocated Python set() objects and intersected them per candidate. The current implementation uses a reusable bytearray with one mark byte per kept pHash:

# Partition 0
candidate_marks[17] = 1
candidate_marks[31] = 1
candidate_marks[44] = 1
candidate_marks[91] = 1

# Partition 1
if candidate_marks[17] == 1: exact_check(17)
if candidate_marks[88] == 1: exact_check(88)  # skipped; mark is zero
if candidate_marks[91] == 1: exact_check(91)

After each query, only touched indexes are cleared. This computes the same intersection while avoiding large temporary set allocations.

Real Dataset Profile

I profiled this on an internal real production dataset with 278,587 rows. Because this repository is public, the dataset identifier and name are intentionally omitted from the PR body. The API key was not included in the report or the PR.

The profiler exported rows with items_and_annotation_generator(), held the exported rows in memory, and then timed only deduplicate_by_phash(...) so export/API time and local dedup CPU time are separated.

Local dedup timing only:

The practical read: the expensive exact local dedup case took about 6m 36.6s for 278,587 real rows. If a user exports the dataset and runs threshold 10, the observed end-to-end time for this dataset would be roughly 127.04s + 396.58s = 523.62s, or about 8m 43.6s.

That is a reasonable runtime for an offline local utility, especially because the implementation is deterministic, exact, requires no server-side job, and validates pHashes before starting the expensive pass.

Validation

• NUCLEUS_PYTEST_API_KEY=dummy .venv/bin/python -m pytest tests/test_local_deduplication.py -q
• Real dataset profile using the local SDK editable install in .venv.

Focused tests cover row preservation, DatasetItem inputs, threshold 0, threshold 64, invalid thresholds, malformed pHashes, unsupported input shapes, custom sort keys, and parity against a simple linear baseline for thresholds 1, 5, and 10.

Greptile Summary

This PR adds a fully offline deduplicate_by_phash() utility and LocalDeduplicationResult to the Nucleus SDK. It requires no API key, accepts DatasetItem objects or generator rows, and dispatches to an exact-match path (threshold 0), a two-partition chunked Hamming index (thresholds 1–10), a linear greedy fallback (thresholds 11–63), or a keep-one path (threshold 64).

• nucleus/local_deduplication.py: New module implementing the full pipeline — pHash validation, stable-id tie-breaking via _tie_break_key, the _HammingIndex with pigeonhole-based candidate filtering and reusable bytearray marks, and the three dedup paths.
• tests/test_local_deduplication.py: Covers all dispatch paths, custom/integer/None sort keys, ordinal fallback, missing reference IDs, and parity against a linear baseline for thresholds 1, 5, 10, 11, and 20.
• nucleus/__init__.py: Re-exports deduplicate_by_phash and LocalDeduplicationResult from the public namespace.

Important Files Changed

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[deduplicate_by_phash] --> B[_validate_threshold]
    B --> C[_normalize_records]
    C --> D{empty?}
    D -- yes --> E[return empty result]
    D -- no --> F[sort by phash_value, stable_id]
    F --> G{threshold?}
    G -- 0 --> H[_deduplicate_exact]
    G -- 64 --> I[keep records 0]
    G -- 1..10 --> J[_deduplicate_with_index]
    G -- 11..63 --> K[_deduplicate_with_linear_scan]
    H --> L[LocalDeduplicationResult]
    I --> L
    J --> L
    K --> L

Fix the following 1 code review issue. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 1
nucleus/local_deduplication.py:162
The `sort_key` callable's return type is annotated as `Union[str, int]`, but `_tie_break_key` explicitly handles a `None` return, and `test_deduplicate_by_phash_preserves_ordinal_fallback_order` relies on it with `sort_key=lambda row: None`. A caller reading the signature has no indication that returning `None` is valid and falls back to ordinal ordering — the annotation should reflect this.

```suggestion
    sort_key: Optional[Callable[[InputT], Optional[Union[str, int]]]] = None,
```

_{Reviews (8): Last reviewed commit: "Run pre-commit formatting" | Re-trigger Greptile}