在 ray_dataset.py 里，算子通过 dataset.map_batches(op.compute_stats, ...) 分发到各个 worker。Ray 在调度时会对 op.compute_stats（bound method）进行 pickle 序列化，而此时 op.backend._dedup_sets = None（因为懒初始化，还没有创建 actor）。每个 worker 进程都独立创建自己的一套 actor，各 worker 的 DedupSet 互相隔离
实际后果： 不同 worker 处理的 batch 之间无法共享去重状态，跨 worker 的重复文档无法被检出，导致去重不完整。
在 ray_dataset.py 里，算子通过 dataset.map_batches(op.compute_stats, ...) 分发到各个 worker。Ray 在调度时会对 op.compute_stats（bound method）进行 pickle 序列化，而此时 op.backend._dedup_sets = None（因为懒初始化，还没有创建 actor）。每个 worker 进程都独立创建自己的一套 actor，各 worker 的 DedupSet 互相隔离
实际后果： 不同 worker 处理的 batch 之间无法共享去重状态，跨 worker 的重复文档无法被检出，导致去重不完整。