Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cmgzn wants to merge 1 commit into
base: main
Choose a base branch
from
+84 −0
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions data_juicer/ops/deduplicator/document_minhash_deduplicator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,9 @@ def process(self, dataset, show_num=0):
if len(dataset) <= 1:
return dataset, {}

# Reset hash tables to avoid stale state from previous calls
self.hash_tables = [defaultdict(set) for _ in range(self.num_bands)]

Comment on lines +298 to +300

@fengrui-z fengrui-z Aug 15, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

minhashes = dataset[HashKeys.minhash]
# remove bytes minhash column otherwise unexpected error would occur
# when exporting the processed dataset
Expand Down Expand Up @@ -385,6 +388,9 @@ def process(self, dataset, show_num=0):
if len(dataset) <= 1:
return dataset, {}

# Reset hash tables to avoid stale state from previous calls
self.hash_tables = [defaultdict(set) for _ in range(self.num_bands)]

@fengrui-z fengrui-z Aug 15, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

minhashes = dataset[HashKeys.minhash]
# remove bytes minhash column otherwise unexpected error would occur
# when exporting the processed dataset
Expand Down
78 changes: 78 additions & 0 deletions tests/ops/deduplicator/test_document_minhash_deduplicator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -977,3 +977,81 @@ def test_chinese_deduplication(self):

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


class DocumentMinhashDeduplicatorRepeatedCallTest(DataJuicerTestCaseBase):

@fengrui-z fengrui-z Aug 15, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

"""Regression test: process() must not leak state between calls."""

def test_repeated_process_no_pollution(self):
"""Calling process() twice on the same operator with disjoint datasets
must produce identical results to calling it once on each dataset
independently."""
from data_juicer.ops.deduplicator.document_minhash_deduplicator import (
DocumentMinhashDeduplicator,
)

# Two completely unrelated datasets
ds_a = Dataset.from_list([
{'text': 'The quick brown fox jumps over the lazy dog'},
{'text': 'A completely unique sentence about quantum physics'},
])
ds_b = Dataset.from_list([
{'text': 'Pack my box with five dozen liquor jugs'},
{'text': 'How vexingly quick daft zebras jump'},
])

op = DocumentMinhashDeduplicator(
tokenization='space',
num_permutations=128,
jaccard_threshold=0.7,
)

# First call
ds_a_hashed = ds_a.map(op.compute_hash)
result_a, _ = op.process(ds_a_hashed)

# Second call on same operator instance — must not be affected by first
ds_b_hashed = ds_b.map(op.compute_hash)
result_b, _ = op.process(ds_b_hashed)

# Neither dataset has duplicates, so both should retain all samples
self.assertEqual(len(result_a), 2,
"First process() call should keep both distinct samples")
self.assertEqual(len(result_b), 2,
"Second process() call should not be polluted by the first")

def test_repeated_process_with_uid_no_pollution(self):
"""Same test for the UID-based variant."""
from data_juicer.ops.deduplicator.document_minhash_deduplicator import (
DocumentMinhashDeduplicatorWithUid,
)
from data_juicer.utils.constant import HashKeys

ds_a = Dataset.from_list([
{'text': 'The quick brown fox jumps over the lazy dog',
HashKeys.uid: 0},
{'text': 'A completely unique sentence about quantum physics',
HashKeys.uid: 1},
])
ds_b = Dataset.from_list([
{'text': 'Pack my box with five dozen liquor jugs',
HashKeys.uid: 0},
{'text': 'How vexingly quick daft zebras jump',
HashKeys.uid: 1},
])

op = DocumentMinhashDeduplicatorWithUid(
tokenization='space',
num_permutations=128,
jaccard_threshold=0.7,
)

ds_a_hashed = ds_a.map(op.compute_hash)
result_a, _ = op.process(ds_a_hashed)

ds_b_hashed = ds_b.map(op.compute_hash)
result_b, _ = op.process(ds_b_hashed)

self.assertEqual(len(result_a), 2)
self.assertEqual(len(result_b), 2,
"UID-based dedup must not carry state between process() calls")
Loading