DGraph

DGraph

DGraph(data: 'DGData', device: str | device = 'cpu')

Dynamic Graph object providing a view over a DGStorage backend.

This class allows efficient slicing, batching, and materialization of temporal/dynamic graph data. It exposes properties for node and edge counts, features, timestamps, and supports device placement for tensors.

Parameters:

• data (DGData) –

  The source DGData object to construct the dynamic graph view.

• device (str | device, default: 'cpu' ) –

  The device to place tensors on. Defaults to 'cpu'.

Raises:

• TypeError –

  If data is not a DGData instance.

Note

• Slicing operations (slice_events or slice_time) return a new DGraph view sharing the underlying storage; they do not copy data unless explicitly materialized via materialize().
• Cached properties (e.g., num_nodes, edges, static_node_x) are computed on first access and then stored. Modifying the underlying storage does not automatically invalidate these cached values.
• materialize() returns dense tensors for src, dst, time, and optionally dynamic node and edge features. This operation may be memory-intensive for large graphs.
• slice_events uses event indices (position in chronological order), while slice_time uses timestamp values. For slice_time, the end_time is exclusive but internally adjusted to include events at end_time - 1.
• num_nodes counts the maximum node ID in the slice + 1; this may differ from the number of nodes in the underlying DGData if slicing excludes some nodes.
• static_node_x shape is (num_nodes_global, d_node_static) and is independent of slices, whereas node_x reflects the current slice.
• Operations such as .to(device) or slicing create views; data is not copied unless explicitly materialized.

Methods:

• materialize –

  Materialize the current DGraph slice into a dense DGBatch.

• num_edge_events –

  The total number of edge events in the dynamic graph.

• num_events –

  The total number of events encountered over the dynamic graph.

• num_node_events –

  The total number of node events in the dynamic graph.

• num_node_labels –

  The total number of node labels in the dynamic graph.

• num_nodes –

  The total number of unique nodes encountered over the dynamic graph.

• num_timestamps –

  The total number of unique timestamps encountered over the dynamic graph.

• slice_events –

  Return a new DGraph view sliced by event indices (end_idx exclusive).

• slice_time –

  Return a new DGraph view sliced by timestamps (end_time exclusive).

• to –

  Return a copy of the DGraph view on a different device.

Attributes:

• edge_dst (Tensor) –

  The edge dst tensor over the dynamic graph.

• edge_src (Tensor) –

  The edge src tensor over the dynamic graph.

• edge_time (Tensor) –

  The timestamps associated with the edge events over the dynamic graph.

• edge_type (Optional[Tensor]) –

  The aggregated edge type over the dynamic graph.

• edge_x (Optional[Tensor]) –

  The aggregated edge features over the dynamic graph.

• edge_x_dim (Optional[int]) –

  Edge feature dimension or None if not Node features on the Graph.

• end_time (Optional[int]) –

  The end time of the dynamic graph. None, if the graph is empty.

• node_type (Optional[Tensor]) –

  If node types exist, returns a dense Tensor(num_nodes_global).

• node_x (Optional[Tensor]) –

  The aggregated dynamic node features over the dynamic graph.

• node_x_dim (Optional[int]) –

  Dynamic Node feature dimension or None if not Node features on the Graph.

• node_x_nids (Tensor) –

  The node ids for associated with the node events over the dynamic graph.

• node_x_time (Tensor) –

  The timestamps associated with the node events over the dynamic graph.

• node_y (Optional[Tensor]) –

  The aggregated dynamic node labels over the dynamic graph.

• node_y_dim (Optional[int]) –

  Dynamic Node label feature dimension or None if no Node labels on the Graph.

• node_y_nids (Tensor) –

  The node ids for associated with the node labels over the dynamic graph.

• node_y_time (Tensor) –

  The timestamps associated with the node labels over the dynamic graph.

• start_time (Optional[int]) –

  The start time of the dynamic graph. None if the graph is empty.

• static_node_x (Optional[Tensor]) –

  If static node features exist, returns a dense Tensor(num_nodes_global x d_node_static).

• static_node_x_dim (Optional[int]) –

  Static Node feature dimension or None if not Node features on the Graph.

Source code in tgm/core/graph.py

def __init__(self, data: 'DGData', device: str | torch.device = 'cpu') -> None:  # type: ignore
    from tgm.data import DGData  # Avoid circular dependency

    if not isinstance(data, DGData):
        raise TypeError(f'DGraph must be initialized with DGData, got {type(data)}')

    self._time_delta = data.time_delta
    self._storage = DGStorage(data)
    self._device = torch.device(device)
    self._slice = DGSliceTracker()

    logger.debug(
        'Created DGraph with device=%s, time_delta=%s, device, data.time_delta'
    )

edge_dst property

edge_dst: Tensor

The edge dst tensor over the dynamic graph.

edge_src property

edge_src: Tensor

The edge src tensor over the dynamic graph.

edge_time property

edge_time: Tensor

The timestamps associated with the edge events over the dynamic graph.

edge_type property

edge_type: Optional[Tensor]

The aggregated edge type over the dynamic graph.

If edge type exist, returns a tensor of shape (T x V x V).

edge_x property

edge_x: Optional[Tensor]

The aggregated edge features over the dynamic graph.

If edge features exist, returns a tensor of shape (T x V x V x d_edge).

edge_x_dim cached property

edge_x_dim: Optional[int]

Edge feature dimension or None if not Node features on the Graph.

end_time cached property

end_time: Optional[int]

The end time of the dynamic graph. None, if the graph is empty.

node_type property

node_type: Optional[Tensor]

If node types exist, returns a dense Tensor(num_nodes_global).

Note

• num_nodes_global is the global number of nodes from the underlying DGData and it will be independent of the slice.

node_x property

node_x: Optional[Tensor]

The aggregated dynamic node features over the dynamic graph.

If dynamic node features exist, returns a Tensor.sparse_coo_tensor(T x V x d_node_dynamic).

node_x_dim cached property

node_x_dim: Optional[int]

Dynamic Node feature dimension or None if not Node features on the Graph.

node_x_nids property

node_x_nids: Tensor

The node ids for associated with the node events over the dynamic graph.

node_x_time property

node_x_time: Tensor

The timestamps associated with the node events over the dynamic graph.

node_y property

node_y: Optional[Tensor]

The aggregated dynamic node labels over the dynamic graph.

If dynamic node labels exist, returns a Tensor.sparse_coo_tensor(T x V x d_node_label).

node_y_dim cached property

node_y_dim: Optional[int]

Dynamic Node label feature dimension or None if no Node labels on the Graph.

node_y_nids property

node_y_nids: Tensor

The node ids for associated with the node labels over the dynamic graph.

node_y_time property

node_y_time: Tensor

The timestamps associated with the node labels over the dynamic graph.

start_time cached property

start_time: Optional[int]

The start time of the dynamic graph. None if the graph is empty.

static_node_x property

static_node_x: Optional[Tensor]

If static node features exist, returns a dense Tensor(num_nodes_global x d_node_static).

Note

• num_nodes_global is the global number of nodes from the underlying DGData and it will be independent of the slice.

static_node_x_dim cached property

static_node_x_dim: Optional[int]

Static Node feature dimension or None if not Node features on the Graph.

materialize

materialize(materialize_features: bool = True) -> DGBatch

Materialize the current DGraph slice into a dense DGBatch.

Parameters:

• materialize_features (bool, default: True ) –

  If True, includes dynamic node features, node IDs/times, and edge features. Defaults to True.

Returns:

• DGBatch ( DGBatch ) –

  A batch containing src, dst, timestamps, and optionally features from the current slice.

Source code in tgm/core/graph.py

@log_latency(level=logging.DEBUG)
def materialize(self, materialize_features: bool = True) -> DGBatch:
    """Materialize the current DGraph slice into a dense `DGBatch`.

    Args:
        materialize_features (bool, optional): If True, includes dynamic node
            features, node IDs/times, and edge features. Defaults to True.

    Returns:
        DGBatch: A batch containing src, dst, timestamps, and optionally
            features from the current slice.
    """
    batch = DGBatch(self.edge_src, self.edge_dst, self.edge_time)
    if materialize_features and self.node_x is not None:
        batch.node_x_time, batch.node_x_nids = self.node_x._indices()
        batch.node_x_nids = batch.node_x_nids.to(torch.int32)  # type: ignore
        batch.node_x = self.node_x._values()

    if materialize_features and self.edge_x is not None:
        batch.edge_x = self.edge_x

    if materialize_features and self.node_y is not None:
        batch.node_y_time, batch.node_y_nids = self.node_y._indices()
        batch.node_y_nids = batch.node_y_nids.to(torch.int32)  # type: ignore
        batch.node_y = self.node_y._values()

    if self.edge_type is not None:
        batch.edge_type = self.edge_type

    logger.debug(
        'Materialized DGraph slice: %d edge events, %d node events, %d node labels',
        batch.edge_src.numel(),
        0 if batch.node_x_nids is None else batch.node_x_nids.numel(),
        0 if batch.node_y_nids is None else batch.node_y_nids.numel(),
    )
    return batch

num_edge_events

num_edge_events() -> int

The total number of edge events in the dynamic graph.

Source code in tgm/core/graph.py

@_logged_cached_property
def num_edge_events(self) -> int:
    """The total number of edge events in the dynamic graph."""
    return len(self.edge_time)

num_events

num_events() -> int

The total number of events encountered over the dynamic graph.

Source code in tgm/core/graph.py

@_logged_cached_property
def num_events(self) -> int:
    """The total number of events encountered over the dynamic graph."""
    return self._storage.get_num_events(self._slice)

num_node_events

num_node_events() -> int

The total number of node events in the dynamic graph.

Source code in tgm/core/graph.py

@_logged_cached_property
def num_node_events(self) -> int:
    """The total number of node events in the dynamic graph."""
    return len(self.node_x_time)

num_node_labels

num_node_labels() -> int

The total number of node labels in the dynamic graph.

Source code in tgm/core/graph.py

@_logged_cached_property
def num_node_labels(self) -> int:
    """The total number of node labels in the dynamic graph."""
    return len(self.node_y_time)

num_nodes

num_nodes() -> int

The total number of unique nodes encountered over the dynamic graph.

Source code in tgm/core/graph.py

@_logged_cached_property
def num_nodes(self) -> int:
    """The total number of unique nodes encountered over the dynamic graph."""
    nodes = self._storage.get_nodes(self._slice)
    return max(nodes) + 1 if len(nodes) else 0

num_timestamps

num_timestamps() -> int

The total number of unique timestamps encountered over the dynamic graph.

Source code in tgm/core/graph.py

@_logged_cached_property
def num_timestamps(self) -> int:
    """The total number of unique timestamps encountered over the dynamic graph."""
    return self._storage.get_num_timestamps(self._slice)

slice_events

slice_events(
    start_idx: Optional[int] = None,
    end_idx: Optional[int] = None,
) -> DGraph

Return a new DGraph view sliced by event indices (end_idx exclusive).

Parameters:

• start_idx (int, default: None ) –

  Starting event index.

• end_idx (int, default: None ) –

  Ending event index (exclusive).

Raises:

• ValueError –

  If start_idx > end_idx.

Source code in tgm/core/graph.py

def slice_events(
    self, start_idx: Optional[int] = None, end_idx: Optional[int] = None
) -> DGraph:
    """Return a new DGraph view sliced by event indices (end_idx exclusive).

    Args:
        start_idx (int, optional): Starting event index.
        end_idx (int, optional): Ending event index (exclusive).

    Raises:
        ValueError: If start_idx > end_idx.
    """
    if start_idx is not None and end_idx is not None and start_idx > end_idx:
        raise ValueError(f'start_idx ({start_idx}) must be <= end_idx ({end_idx})')

    slice = replace(self._slice)
    slice.start_idx = self._maybe_max(start_idx, slice.start_idx)
    slice.end_idx = self._maybe_min(end_idx, slice.end_idx)
    return DGraph._from_storage(self._storage, self.time_delta, self.device, slice)

slice_time

slice_time(
    start_time: Optional[int] = None,
    end_time: Optional[int] = None,
) -> DGraph

Return a new DGraph view sliced by timestamps (end_time exclusive).

Parameters:

• start_time (int, default: None ) –

  Starting timestamp.

• end_time (int, default: None ) –

  Ending timestamp (exclusive).

Raises:

• ValueError –

  If start_time > end_time.

Source code in tgm/core/graph.py

def slice_time(
    self, start_time: Optional[int] = None, end_time: Optional[int] = None
) -> DGraph:
    """Return a new DGraph view sliced by timestamps (end_time exclusive).

    Args:
        start_time (int, optional): Starting timestamp.
        end_time (int, optional): Ending timestamp (exclusive).

    Raises:
        ValueError: If start_time > end_time.
    """
    if start_time is not None and end_time is not None and start_time > end_time:
        raise ValueError(
            f'start_time ({start_time}) must be <= end_time ({end_time})'
        )
    if end_time is not None:
        end_time -= 1

    slice = replace(self._slice)
    slice.start_time = self._maybe_max(start_time, slice.start_time)
    slice.end_time = self._maybe_min(end_time, slice.end_time)
    return DGraph._from_storage(self._storage, self.time_delta, self.device, slice)

to

to(device: str | device) -> DGraph

Return a copy of the DGraph view on a different device.

Parameters:

• device (str | device) –

  The target device.

Returns:

• DGraph ( DGraph ) –

  A new view on the specified device.

Source code in tgm/core/graph.py

def to(self, device: str | torch.device) -> DGraph:
    """Return a copy of the DGraph view on a different device.

    Args:
        device (str | torch.device): The target device.

    Returns:
        DGraph: A new view on the specified device.
    """
    logger.debug('Moving DGraph to device %s', device)
    device = torch.device(device)
    slice = replace(self._slice)
    return DGraph._from_storage(self._storage, self.time_delta, device, slice)