fix(deduplicator): reset hash_tables between process() calls

[cmgzn]

Reproduction

The same 3-sample dataset processed with the same parameters gives different deduplication results depending on whether the operator was previously used on other data.

from datasets import Dataset
from data_juicer.ops.deduplicator.document_minhash_deduplicator import (
    DocumentMinhashDeduplicator,
)

target = Dataset.from_list([
    {"text": "the cat sat on the mat by the door"},
    {"text": "the dog sat on the rug by the window"},
    {"text": "a fish swam in the tank near the light"},
])

op = DocumentMinhashDeduplicator(
    tokenization="space", num_permutations=16, jaccard_threshold=0.5,
)

# Process some other data first (simulating a probe/preview run)
warmup = Dataset.from_list([
    {"text": "the cat sat on the mat by the door today"},
    {"text": "the cat lay on the mat by the wall yesterday"},
    {"text": "the dog sat on the rug by the window again"},
    {"text": "the dog lay on the rug by the curtain now"},
    {"text": "some unrelated text about cooking pasta sauce"},
    {"text": "another sentence about hiking in the mountains"},
])
op.process(warmup.map(op.compute_hash))

# Now process our target dataset on the same operator instance
result, _ = op.process(target.map(op.compute_hash))
print(f"Reused operator: {len(result)}/3 retained")

# Compare with a fresh operator
op_fresh = DocumentMinhashDeduplicator(
    tokenization="space", num_permutations=16, jaccard_threshold=0.5,
)
result_fresh, _ = op_fresh.process(target.map(op_fresh.compute_hash))
print(f"Fresh operator:  {len(result_fresh)}/3 retained")

Before fix

Reused operator: 2/3 retained
Fresh operator:  3/3 retained

Same input, same parameters — 1 sample falsely deduplicated because stale hash-table entries from the warmup run caused a spurious UnionFind merge.

After fix

Reused operator: 3/3 retained
Fresh operator:  3/3 retained

Results are identical regardless of operator history.

Root cause

self.hash_tables (list of defaultdict(set)) is created once in __init__ and populated with sample indices inside process(), but never cleared between calls. Index entries from a prior process() call persist and participate in clustering on subsequent calls, merging unrelated samples.

Fix

Reset self.hash_tables at the top of process():

self.hash_tables = [defaultdict(set) for _ in range(self.num_bands)]

Applied to both DocumentMinhashDeduplicator and DocumentMinhashDeduplicatorWithUid.

Tests added

DocumentMinhashDeduplicatorRepeatedCallTest — calls process() twice on the same operator instance with disjoint datasets, asserts no cross-contamination. Covers both the base class and the UID variant.

[fengrui-z]

If process() is called with an empty or single-sample dataset, hash_tables from the previous call remains populated on the instance. The next large call does reset before use, so output is currently safe, but this breaks the stated invariant ("reset at the top of process()") and leaves stale state observable after a small/empty call.

Please move the reset before the early return, ideally with a shared helper:

def _reset_hash_tables(self):
    self.hash_tables = [defaultdict(set) for _ in range(self.num_bands)]
and call self._reset_hash_tables() as the first statement in both 

process() implementations.

[fengrui-z]

The new test class is defined after the main guard:

if __name__ == '__main__':
    unittest.main()

class DocumentMinhashDeduplicatorRepeatedCallTest(...):
    ...

python test_document_minhash_deduplicator.py will run unittest.main() and exit before the class definition is ever executed, so the new tests are not discoverable when the file is run directly. Please move the class above line 978, or move the if __name__ == '__main__' guard to the end of the file.

[fengrui-z]

Same issue as above for DocumentMinhashDeduplicatorWithUid.process(): reset at line 391 should occur before the if len(dataset) <= 1 early return at lines 388–389. If you add _reset_hash_tables() to the base class, both overrides can simply call it first.