From 17e907f2e9cd5645aec23d8965f4d7faaa8e7de7 Mon Sep 17 00:00:00 2001
From: pulp <chemgzn@163.com>
Date: Fri, 14 Aug 2026 17:04:28 +0800
Subject: [PATCH] fix(deduplicator): reset hash_tables between process() calls

hash_tables was initialized once in __init__ but never cleared when
process() was called again on the same operator instance. A prior call
(e.g. probe/preview) would leave stale entries that pollute subsequent
deduplication, causing false-positive merges via UnionFind.

Part of cmgzn/data-juicer#137
---
 .../document_minhash_deduplicator.py          |  6 ++
 .../test_document_minhash_deduplicator.py     | 78 +++++++++++++++++++
 2 files changed, 84 insertions(+)

diff --git a/data_juicer/ops/deduplicator/document_minhash_deduplicator.py b/data_juicer/ops/deduplicator/document_minhash_deduplicator.py
index 3946f8d60ca..53bd744b67c 100644
--- a/data_juicer/ops/deduplicator/document_minhash_deduplicator.py
+++ b/data_juicer/ops/deduplicator/document_minhash_deduplicator.py
@@ -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)]
+
         minhashes = dataset[HashKeys.minhash]
         # remove bytes minhash column otherwise unexpected error would occur
         # when exporting the processed dataset
@@ -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)]
+
         minhashes = dataset[HashKeys.minhash]
         # remove bytes minhash column otherwise unexpected error would occur
         # when exporting the processed dataset
diff --git a/tests/ops/deduplicator/test_document_minhash_deduplicator.py b/tests/ops/deduplicator/test_document_minhash_deduplicator.py
index e335da2de24..83056d6d59e 100644
--- a/tests/ops/deduplicator/test_document_minhash_deduplicator.py
+++ b/tests/ops/deduplicator/test_document_minhash_deduplicator.py
@@ -977,3 +977,81 @@ def test_chinese_deduplication(self):
 
 if __name__ == '__main__':
     unittest.main()
+
+
+class DocumentMinhashDeduplicatorRepeatedCallTest(DataJuicerTestCaseBase):
+    """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")