def METHOD_NAME(self):
    from .fetch import FetchShuffle
    if self.stage == OperandStage.reduce:
        inputs = self.inputs or ()
        deps = []
        for inp in inputs:
            if isinstance(inp.op, ShuffleProxy):
                deps.extend(
                    [(chunk.key, self.reducer_index) for chunk in inp.inputs or ()]
                )
            elif isinstance(inp.op, FetchShuffle):
                # fetch shuffle by index doesn't store data keys, so it won't run into this function.
                assert inp.op.shuffle_fetch_type == ShuffleFetchType.FETCH_BY_KEY
                deps.extend([(k, self.reducer_index) for k in inp.op.source_keys])
            else:
                deps.append(inp.key)
        return deps
    return super().METHOD_NAME()