def _seeds_preprocess(
        minibatch: MiniBatch,
        rank: int = 0,
        world_size: int = 1,
        async_op: bool = False,
    ):
        """Preprocess `seeds` in a minibatch to construct `unique_seeds`,
        `node_timestamp` and `compacted_seeds` for further sampling. It
        optionally incorporates timestamps for temporal graphs, organizing and
        compacting seeds based on their types and timestamps. In heterogeneous
        graph, `seeds` with same node type will be unqiued together.

        Parameters
        ----------
        minibatch: MiniBatch
            The minibatch.
        rank : int
            The rank of the current process among cooperating processes.
        world_size : int
            The number of cooperating
            (`arXiv:2210.13339<https://arxiv.org/abs/2310.12403>`__) processes.
        async_op: bool
            Boolean indicating whether the call is asynchronous. If so, the
            result can be obtained by calling wait on the returned future.

        Returns
        -------
        unique_seeds: torch.Tensor or Dict[str, torch.Tensor]
            A tensor or a dictionary of tensors representing the unique seeds.
            In heterogeneous graphs, seeds are returned for each node type.
        nodes_timestamp: None or a torch.Tensor or Dict[str, torch.Tensor]
            Containing timestamps for each seed. This is only returned if
            `minibatch` includes timestamps and the graph is temporal.
        compacted_seeds: torch.tensor or a Dict[str, torch.Tensor]
            Representation of compacted seeds corresponding to 'seeds', where
            all node ids inside are compacted.
        offsets: None or torch.Tensor or Dict[src, torch.Tensor]
            The unique nodes offsets tensor partitions the unique_nodes tensor.
            Has size `world_size + 1` and
            `unique_nodes[offsets[i]: offsets[i + 1]]` belongs to the rank
            `(rank + i) % world_size`.
        """
        use_timestamp = hasattr(minibatch, "timestamp")
        assert (
            not use_timestamp or world_size == 1
        ), "Temporal code path does not currently support Cooperative Minibatching"
        seeds = minibatch.seeds
        is_heterogeneous = isinstance(seeds, Dict)
        if is_heterogeneous:
            # Collect nodes from all types of input.
            nodes = defaultdict(list)
            nodes_timestamp = None
            if use_timestamp:
                nodes_timestamp = defaultdict(list)
            for seed_type, typed_seeds in seeds.items():
                # When typed_seeds is a one-dimensional tensor, it represents
                # seed nodes, which does not need to do unique and compact.
                if typed_seeds.ndim == 1:
                    nodes_timestamp = (
                        minibatch.timestamp
                        if hasattr(minibatch, "timestamp")
                        else None
                    )
                    result = _NoOpWaiter((seeds, nodes_timestamp, None, None))
                    break
                result = None
                assert typed_seeds.ndim == 2, (
                    "Only tensor with shape 1*N and N*M is "
                    + f"supported now, but got {typed_seeds.shape}."
                )
                ntypes = seed_type_str_to_ntypes(
                    seed_type, typed_seeds.shape[1]
                )
                if use_timestamp:
                    negative_ratio = (
                        typed_seeds.shape[0]
                        // minibatch.timestamp[seed_type].shape[0]
                        - 1
                    )
                    neg_timestamp = minibatch.timestamp[
                        seed_type
                    ].repeat_interleave(negative_ratio)
                for i, ntype in enumerate(ntypes):
                    nodes[ntype].append(typed_seeds[:, i])
                    if use_timestamp:
                        nodes_timestamp[ntype].append(
                            minibatch.timestamp[seed_type]
                        )
                        nodes_timestamp[ntype].append(neg_timestamp)

            class _Waiter:
                def __init__(self, nodes, nodes_timestamp, seeds):
                    # Unique and compact the collected nodes.
                    if use_timestamp:
                        self.future = compact_temporal_nodes(
                            nodes, nodes_timestamp
                        )
                    else:
                        self.future = unique_and_compact(
                            nodes, rank, world_size, async_op
                        )
                    self.seeds = seeds

                def wait(self):
                    """Returns the stored value when invoked."""
                    if use_timestamp:
                        unique_seeds, nodes_timestamp, compacted = self.future
                        offsets = None
                    else:
                        unique_seeds, compacted, offsets = (
                            self.future.wait() if async_op else self.future
                        )
                        nodes_timestamp = None
                    seeds = self.seeds
                    # Ensure there is no memory leak.
                    self.future = self.seeds = None

                    compacted_seeds = {}
                    # Map back in same order as collect.
                    for seed_type, typed_seeds in seeds.items():
                        ntypes = seed_type_str_to_ntypes(
                            seed_type, typed_seeds.shape[1]
                        )
                        compacted_seed = []
                        for ntype in ntypes:
                            compacted_seed.append(compacted[ntype].pop(0))
                        compacted_seeds[seed_type] = (
                            torch.cat(compacted_seed).view(len(ntypes), -1).T
                        )

                    return (
                        unique_seeds,
                        nodes_timestamp,
                        compacted_seeds,
                        offsets,
                    )

            # When typed_seeds is not a one-dimensional tensor
            if result is None:
                result = _Waiter(nodes, nodes_timestamp, seeds)
        else:
            # When seeds is a one-dimensional tensor, it represents seed nodes,
            # which does not need to do unique and compact.
            if seeds.ndim == 1:
                nodes_timestamp = (
                    minibatch.timestamp
                    if hasattr(minibatch, "timestamp")
                    else None
                )
                result = _NoOpWaiter((seeds, nodes_timestamp, None, None))
            else:
                # Collect nodes from all types of input.
                nodes = [seeds.view(-1)]
                nodes_timestamp = None
                if use_timestamp:
                    # Timestamp for source and destination nodes are the same.
                    negative_ratio = (
                        seeds.shape[0] // minibatch.timestamp.shape[0] - 1
                    )
                    neg_timestamp = minibatch.timestamp.repeat_interleave(
                        negative_ratio
                    )
                    seeds_timestamp = torch.cat(
                        (minibatch.timestamp, neg_timestamp)
                    )
                    nodes_timestamp = [
                        seeds_timestamp for _ in range(seeds.shape[1])
                    ]

                class _Waiter:
                    def __init__(self, nodes, nodes_timestamp, seeds):
                        # Unique and compact the collected nodes.
                        if use_timestamp:
                            self.future = compact_temporal_nodes(
                                nodes, nodes_timestamp
                            )
                        else:
                            self.future = unique_and_compact(
                                nodes, async_op=async_op
                            )
                        self.seeds = seeds

                    def wait(self):
                        """Returns the stored value when invoked."""
                        if use_timestamp:
                            (
                                unique_seeds,
                                nodes_timestamp,
                                compacted,
                            ) = self.future
                            offsets = None
                        else:
                            unique_seeds, compacted, offsets = (
                                self.future.wait() if async_op else self.future
                            )
                            nodes_timestamp = None
                        seeds = self.seeds
                        # Ensure there is no memory leak.
                        self.future = self.seeds = None

                        # Map back in same order as collect.
                        compacted_seeds = compacted[0].view(seeds.shape)

                        return (
                            unique_seeds,
                            nodes_timestamp,
                            compacted_seeds,
                            offsets,
                        )

                result = _Waiter(nodes, nodes_timestamp, seeds)

        return result if async_op else result.wait()
Preprocess `seeds` in a minibatch to construct `unique_seeds`,
        `node_timestamp` and `compacted_seeds` for further sampling. It
        optionally incorporates timestamps for temporal graphs, organizing and
        compacting seeds based on their types and timestamps. In heterogeneous
        graph, `seeds` with same node type will be unqiued together.

        Parameters
        ----------
        minibatch: MiniBatch
            The minibatch.
        rank : int
            The rank of the current process among cooperating processes.
        world_size : int
            The number of cooperating
            (`arXiv:2210.13339<https://arxiv.org/abs/2310.12403>`__) processes.
        async_op: bool
            Boolean indicating whether the call is asynchronous. If so, the
            result can be obtained by calling wait on the returned future.

        Returns
        -------
        unique_seeds: torch.Tensor or Dict[str, torch.Tensor]
            A tensor or a dictionary of tensors representing the unique seeds.
            In heterogeneous graphs, seeds are returned for each node type.
        nodes_timestamp: None or a torch.Tensor or Dict[str, torch.Tensor]
            Containing timestamps for each seed. This is only returned if
            `minibatch` includes timestamps and the graph is temporal.
        compacted_seeds: torch.tensor or a Dict[str, torch.Tensor]
            Representation of compacted seeds corresponding to 'seeds', where
            all node ids inside are compacted.
        offsets: None or torch.Tensor or Dict[src, torch.Tensor]
            The unique nodes offsets tensor partitions the unique_nodes tensor.
            Has size `world_size + 1` and
            `unique_nodes[offsets[i]: offsets[i + 1]]` belongs to the rank
            `(rank + i) % world_size`.
        
python/dgl/graphbolt/subgraph_sampler.py