class_docstr: An implementation of a distributed and asynchronised neighbor sampler
  used by :class:`~torch_geometric.distributed.DistNeighborLoader`.
class_docstr_tok: ["An","implementation","of","a","distributed","and","asynchronised","neighbor","sampler","used","by",":","class",":","`","~torch_geometric.distributed.DistNeighborLoader","`","."]
docstr: Implements one-hop neighbor sampling for a set of input nodes for a
  specific edge type.
docstr_tok: ["Implements","one-hop","neighbor","sampling","for","a","set","of","input","nodes","for","a","specific","edge","type","."]
code: def _sample_one_hop(
      self,
      input_nodes: Tensor,
      num_neighbors: int,
      seed_time: Optional[Tensor] = None,
      edge_type: Optional[EdgeType] = None,
  ) -> SamplerOutput:
      r"""Implements one-hop neighbor sampling for a set of input nodes for a
      specific edge type.
      """
      if not self.is_hetero:
          colptr = self._sampler.colptr
          row = self._sampler.row
          node_time = self.node_time
          edge_time = self.edge_time
      else:
          # Given edge type, get input data and evaluate sample function:
          rel_type = "__".join(edge_type)
          colptr = self._sampler.colptr_dict[rel_type]
          row = self._sampler.row_dict[rel_type]
          # `node_time` is a destination node time:
          node_time = (self.node_time or {}).get(edge_type[0], None)
          edge_time = (self.edge_time or {}).get(edge_type, None)
  
      out = torch.ops.pyg.dist_neighbor_sample(
          colptr,
          row,
          input_nodes.to(colptr.dtype),
          num_neighbors,
          node_time,
          edge_time,
          seed_time,
          None,  # TODO: edge_weight
          True,  # csc
          self.replace,
          self.subgraph_type != SubgraphType.induced,
          self.disjoint and self.temporal,
          self.temporal_strategy,
      )
      node, edge, cumsum_neighbors_per_node = out
  
      if self.disjoint and self.temporal:
          # We create a batch during the step of merging sampler outputs.
          _, node = node.t().contiguous()
  
      return SamplerOutput(
          node=node,
          row=None,
          col=None,
          edge=edge,
          batch=None,
          metadata=(cumsum_neighbors_per_node,),
      )
  
imports: import itertools
  import logging
  import math
  from typing import Any, Callable, Dict, List, Optional, Tuple, Union
  import numpy
  import torch
  import torch.multiprocessing
  from torch import Tensor
  from torch_geometric.distributed import LocalFeatureStore, LocalGraphStore
  from torch_geometric.distributed.dist_context import DistContext
  from torch_geometric.distributed.event_loop import ConcurrentEventLoop, to_asyncio_future
  from torch_geometric.distributed.rpc import RPCCallBase, RPCRouter, rpc_async, rpc_partition_to_workers, rpc_register
  from torch_geometric.distributed.utils import BatchDict, NodeDict, remove_duplicates
  from torch_geometric.sampler import EdgeSamplerInput, HeteroSamplerOutput, NegativeSampling, NeighborSampler, NodeSamplerInput, SamplerOutput
  from torch_geometric.sampler.base import NumNeighbors, SubgraphType
  from torch_geometric.sampler.neighbor_sampler import neg_sample
  from torch_geometric.sampler.utils import remap_keys
  from torch_geometric.typing import EdgeType, NodeType
label_desc: Use image node_id 9 for calling the DistNeighborSampler obj's underlying member method code with example usage: obj._sample_one_hop(input_nodes, num_neighbors, seed_time, edge_type) and returns: SamplerOutput