def _sample_etype_neighbors_dgl(
    local_g,
    partition_book,
    seed_nodes,
    fan_out,
    edge_dir="in",
    prob=None,
    exclude_edges=None,
    replace=False,
    etype_offset=None,
    etype_sorted=False,
):
    """Sample from local partition.

    The input nodes use global IDs. We need to map the global node IDs to local node IDs,
    perform sampling and map the sampled results to the global IDs space again.
    The sampled results are stored in three vectors that store source nodes, destination nodes
    and edge IDs.
    """
    assert etype_offset is not None, "The etype offset is not provided."

    local_ids = partition_book.nid2localnid(seed_nodes, partition_book.partid)
    local_ids = F.astype(local_ids, local_g.idtype)

    sampled_graph = local_sample_etype_neighbors(
        local_g,
        local_ids,
        etype_offset,
        fan_out,
        edge_dir=edge_dir,
        prob=prob,
        exclude_edges=exclude_edges,
        replace=replace,
        etype_sorted=etype_sorted,
        _dist_training=True,
    )
    global_nid_mapping = local_g.ndata[NID]
    src, dst = sampled_graph.edges()
    global_src, global_dst = F.gather_row(
        global_nid_mapping, src
    ), F.gather_row(global_nid_mapping, dst)
    global_eids = F.gather_row(local_g.edata[EID], sampled_graph.edata[EID])
    return LocalSampledGraph(global_src, global_dst, global_eids)
Sample from local partition.

    The input nodes use global IDs. We need to map the global node IDs to local node IDs,
    perform sampling and map the sampled results to the global IDs space again.
    The sampled results are stored in three vectors that store source nodes, destination nodes
    and edge IDs.
    
python/dgl/distributed/graph_services.py