Graph in depth

This page is the full reference for Graph. Start there for the short version.

How the graph works

A graph stores entities as nodes and relationships as edges. laser.graph(name) returns a handle that writes, reads, and traverses one named graph.

A node ID comes from the node's label and value. GraphNode::entity("service", "auth") (TypeScript: graphNodeEntity, Python: graph_node_entity) always produces the same ID in every SDK, so two services that mention service:auth update one node without agreeing on IDs first. An edge ID comes from its two endpoints and its type, so the same relationship recorded twice is one edge. Writes are therefore safe to repeat.

A client with a default stream scopes the graph name to that stream, so laser.graph("kg") addresses stream:<stream>/kg. See Stream-scoped resource names. A graph name is at most 128 bytes with no ASCII control characters.

The graph is a managed feature. It works on Laser Stack and LaserData Cloud when capabilities.graph is true. Against Apache Iggy alone, every graph call returns an unsupported error (LaserError::Unsupported in Rust, UnsupportedError in Python and TypeScript). Graph writes enter a durable mutation log before the deployment applies them, and traversals read with eventual consistency, so a write can take a moment to appear.

Write facts

link(from, relation, to) takes two "label:value" strings. It derives both node IDs, writes both nodes, and writes the edge. A string without a colon gets the label entity. Repeating the same triple produces the same nodes and edge.

upsert(nodes, edges) writes nodes and edges you build yourself, for richer payloads. One upsert carries at most 10,000 nodes and edges, and a node carries at most 16 labels.

A node holds an ID, labels, attributes, an optional embedding, an optional source, and an optional producer. An entity node carries its label in labels and its value in the value attribute. An edge holds an ID, its endpoints, a type, a weight, attributes, an optional valid-time window (valid_from and valid_to in epoch microseconds), a source, and a producer. Attributes are [name, value] pairs in every SDK. Rust and TypeScript return typed structs and objects. Python returns dicts.

Change a fact

These operations keep relationship history:

• Call relink(from, relation, to) to replace a single-valued relationship. It closes every live edge with the same source node and relation that points at another target, writes the new edge, and returns how many edges it closed.
• Call unlink(from, relation, to) to close one edge by setting valid_to to now. The nodes stay. An as_of read at an earlier time still sees the relationship.

Read a neighborhood

neighbors(node_id, direction, edge_type, depth) returns the nodes reachable from one node, with the edges between them. The direction is outgoing, incoming, or both ("out", "in", or "both" in Python and TypeScript, EdgeDir in Rust). An edge type limits the walk to that relationship. Without one, it follows every type. Depth is the number of hops. The reply includes the start node. Python's neighbors(node, dir="out", edge_type=None, depth=1) defaults the last three.

limit, as_of, and conversation on the handle also apply to neighbors.

Traverse

The traversal builder gives more control:

• Start with start_ids([...]), start_match(filter), or start_nearest(embedding, k). These pick explicit nodes, nodes that match a query Filter, or the k nodes closest to an embedding.
• Add hops with out(edge_type), incoming(edge_type), and both(edge_type). Each call adds one hop.
• Return nodes by default, or choose return_edges(), return_triplets(), or return_paths(). Triplets are source, type, destination. Paths are node and edge ID sequences.
• Cap the result with limit(n), 100 by default. as_of(micros) reads edges valid at a time. conversation(id) limits the walk to facts recorded by one conversation. Without it, the walk spans the whole graph.

Finish with fetch(). TypeScript uses camelCase (startIds, returnPaths, asOf). A traversal goes at most 8 hops deep, and a reply holds at most 10,000 nodes and edges. The start nodes are part of the returned nodes.

import { filterPred } from "@laserdata/laser-sdk"

const dependencies = await laser
  .graph("ops")
  .startMatch(filterPred("label", "eq", { kind: "string", value: "Service" }))
  .out("depends_on")
  .limit(100)
  .fetch()

const paths = await laser.graph("ops").startNearest(embedding, 5).out("depends_on").returnPaths().fetch()

let dependencies = laser
    .graph("ops")
    .start_match(Filter::pred("label", CmpOp::Eq, "Service"))
    .out("depends_on")
    .limit(100)
    .fetch()
    .await?;

let paths = laser
    .graph("ops")
    .start_nearest(embedding, 5)
    .out("depends_on")
    .return_paths()
    .fetch()
    .await?;

dependencies = await (
    laser.graph("ops")
    .start_match(ls.Filter.pred("label", "eq", "Service"))
    .out("depends_on")
    .limit(100)
    .fetch()
)

paths = await (
    laser.graph("ops")
    .start_nearest(embedding, 5)
    .out("depends_on")
    .return_paths()
    .fetch()
)

Rust returns a GraphResult with nodes, edges, and paths. TypeScript returns a GraphResult object with the same fields. Python returns a dict with nodes, edges, and paths, each present when the traversal filled it.

Read the graph at a past time

as_of(micros) (TypeScript: asOf, as a bigint) reads the graph as it was at a time in epoch microseconds. Only edges whose valid-time window contains that time are followed. The window includes valid_from and excludes valid_to, and a missing bound is open, so an edge without a window is always valid. A closed edge stays visible to an earlier as_of read.

This example writes a mitigation edge that became true at a known time, then reads before and after it.

import { graphEdgeRelate, graphEdgeValid, graphNodeEntity } from "@laserdata/laser-sdk"

const since = 1_900_000_000_000_000n
const gateway = graphNodeEntity("Service", "gateway")
const replica = graphNodeEntity("Component", "read-replica")
const edge = graphEdgeValid(graphEdgeRelate(gateway, "mitigated_by", replica), since)
await laser.graph("ops").upsert([gateway, replica], [edge])

const before = await laser.graph("ops").startIds([gateway.id]).out("mitigated_by").asOf(since - 1n).fetch()
const after = await laser.graph("ops").startIds([gateway.id]).out("mitigated_by").asOf(since + 1n).fetch()

let since: u64 = 1_900_000_000_000_000;
let gateway = GraphNode::entity("Service", "gateway");
let replica = GraphNode::entity("Component", "read-replica");
let edge = GraphEdge::relate(&gateway, "mitigated_by", &replica).valid(Some(since), None);
laser
    .graph("ops")
    .upsert(vec![gateway.clone(), replica], vec![edge])
    .await?;

let before = laser
    .graph("ops")
    .start_ids(vec![gateway.id])
    .out("mitigated_by")
    .as_of(since - 1)
    .fetch()
    .await?;
let after = laser
    .graph("ops")
    .start_ids(vec![gateway.id])
    .out("mitigated_by")
    .as_of(since + 1)
    .fetch()
    .await?;

since = 1_900_000_000_000_000
gateway = ls.graph_node_entity("Service", "gateway")
replica = ls.graph_node_entity("Component", "read-replica")
edge = ls.graph_edge_valid(ls.graph_edge_relate(gateway, "mitigated_by", replica), since, None)
await laser.graph("ops").upsert([gateway, replica], [edge])

before = await (
    laser.graph("ops").start_ids([gateway["id"]]).out("mitigated_by").as_of(since - 1).fetch()
)
after = await (
    laser.graph("ops").start_ids([gateway["id"]]).out("mitigated_by").as_of(since + 1).fetch()
)

Build the graph from a topic

You can write the graph directly with link and upsert, or derive it from messages. To derive it, register a graph projection with projections().register_graph(projection) (TypeScript: projections().registerGraph) and an entity schema. Each node rule names a label and a JSON pointer (an RFC 6901 path into the record) to the node value, plus an optional pointer to an embedding. Each edge rule names an edge type, pointers to its two endpoints, and optional pointers to a valid-time window. When the projection is bound to a source topic, the deployment applies the rules to each record and writes the nodes and edges, with no extraction code in your services.

import { ContentType, parseProjectionId } from "@laserdata/laser-sdk"

await laser.projections().registerGraph({
  id: parseProjectionId("ops.v1"),
  name: "ops",
  version: 1,
  kind: { kind: "graph" },
  contentType: ContentType.Json,
  extraction: { fields: [], inlinePayload: false },
  entitySchema: {
    nodes: [
      { label: "Service", valuePointer: "/service" },
      { label: "Component", valuePointer: "/component" }
    ],
    edges: [{ edgeType: "depends_on", fromPointer: "/service", toPointer: "/component" }]
  },
  inlinePayloadDefault: false
})

laser
    .projections()
    .register_graph(
        Projection::builder("ops.v1")
            .name("ops")
            .content_type(ContentType::Json)
            .index_only()
            .graph(EntitySchema {
                nodes: vec![
                    NodeExtract {
                        label: "Service".to_owned(),
                        value_pointer: "/service".to_owned(),
                        embedding_pointer: None,
                    },
                    NodeExtract {
                        label: "Component".to_owned(),
                        value_pointer: "/component".to_owned(),
                        embedding_pointer: None,
                    },
                ],
                edges: vec![EdgeExtract {
                    edge_type: "depends_on".to_owned(),
                    from_pointer: "/service".to_owned(),
                    to_pointer: "/component".to_owned(),
                    valid_from_pointer: None,
                    valid_to_pointer: None,
                }],
            })
            .build(),
    )
    .await?;

await laser.projections().register_graph(
    {
        "id": "ops.v1",
        "name": "ops",
        "version": 1,
        "content_type": "json",
        "extraction": {"fields": [], "inline_payload": False},
        "entity_schema": {
            "nodes": [
                {"label": "Service", "value_pointer": "/service"},
                {"label": "Component", "value_pointer": "/component"},
            ],
            "edges": [
                {"edge_type": "depends_on", "from_pointer": "/service", "to_pointer": "/component"},
            ],
        },
    }
)

Queries and views covers binding a projection to a topic.

Session links

A graph write can name the session that produced it. On a graph handle, in_session(reference) links every upsert to the session, produced_by(producer) stamps a producer name and version, and sourced_from(source) stamps the record the facts came from (TypeScript: inSession, producedBy, sourcedFrom). The producer and source apply only to nodes and edges that do not set their own, and they never change a node or edge ID. A node keeps the source it was first written with, and an edge keeps the latest one.

Inside a session, session.linked_graph(name) (TypeScript: linkedGraph) returns a handle with all three set: the session, the session's agent as the producer, and the record the session acts on as the source. Call acting_on(source) (TypeScript: actingOn) on the session first to name that record. session.graph(name) returns the shared graph without a link.

The server checks a session link like a key-value link. The caller needs its graph grant, session:write on the stream, and send permission on the stream's agent.sessions. A managed deployment then lists each node and edge the session touched. One graph can serve many sessions in the same stream, and each session keeps its own history.

const graph = session.linkedGraph("services")
await graph.link("service:auth", "depends_on", "service:db")

const extracted = laser
  .graph("services")
  .inSession(session.reference())
  .producedBy({ name: "dependency-extractor", version: "3" })

use laser_sdk::wire::graph::ProducerInfo;

let graph = session.linked_graph("services")?;
graph.link("service:auth", "depends_on", "service:db").await?;

let extracted = laser
    .graph("services")
    .in_session(session.reference()?)
    .produced_by(ProducerInfo {
        name: "dependency-extractor".into(),
        version: "3".into(),
    });

graph = session.linked_graph("services")
await graph.link("service:auth", "depends_on", "service:db")

extracted = (
    laser.graph("services")
    .in_session(session.reference())
    .produced_by({"name": "dependency-extractor", "version": "3"})
)

A source names a log record (numeric stream, topic, partition, and offset, plus the topic creation time and conversation when known), a key-value entry, or a memory item. TypeScript writes it as { kind: "message", .. }, { kind: "kv", namespace, key }, or { kind: "memory", id }. Python writes it as a dict keyed by the variant, such as {"Kv": {"namespace": "topology", "key": "gateway"}}.

Errors

Python raises GraphError with the cause in detail. TypeScript throws GraphExecutionError with the cause in detail.

Key operations

Requirements

Use Laser Stack or LaserData Cloud. In Rust, the graph needs the graph feature. The graph example checks capabilities.graph and exits normally when the deployment does not serve it.