risingwavelabs/risingwave · error

graph is not a DAG

Error message

graph is not a DAG

What it means

During topological sorting of the stream fragment graph, all fragments should be consumed (their downstream counts reduced to zero). If `downstream_cnts` still has entries after the traversal, the fragment graph contains a cycle (or unreachable nodes), so the builder bails with 'graph is not a DAG'. A cyclic fragment graph cannot be scheduled.

Source

Thrown at src/meta/src/stream/stream_graph/fragment.rs:2149

        }

        let mut i = 0;
        while let Some(&fragment_id) = topo.get(i) {
            i += 1;
            // Find if we can process more fragments.
            for (upstream_job_id, _) in self.get_upstreams(fragment_id) {
                let downstream_cnt = downstream_cnts.get_mut(&upstream_job_id).unwrap();
                *downstream_cnt -= 1;
                if *downstream_cnt == 0 {
                    downstream_cnts.remove(&upstream_job_id);
                    topo.push(upstream_job_id);
                }
            }
        }

        if !downstream_cnts.is_empty() {
            // There are fragments that are not processed yet.
            bail!("graph is not a DAG");
        }

        Ok(topo)
    }

    /// Seal a [`BuildingFragment`] from the graph into a [`Fragment`], which will be further used
    /// to build actors on the compute nodes and persist into meta store.
    pub(super) fn seal_fragment(
        &self,
        id: GlobalFragmentId,
        distribution: Distribution,
        stream_node: StreamNode,
    ) -> Fragment {
        let building_fragment = self.get_fragment(id).into_building().unwrap();
        let internal_tables = building_fragment.extract_internal_tables();
        let BuildingFragment {
            inner,
            job_id,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the fragment graph for the job (dump fragment dependencies) and find the cycle; fix the edge-building code that introduced it.
2. Recreate the streaming job so a fresh, acyclic graph is built.
3. If metadata corruption is suspected, drop and rebuild the job from source DDL rather than repairing fragments.

Example fix

null

// Verify acyclicity of the fragment dependency graph before topo-sort
fn is_acyclic(nodes: &[FragmentId], edges: &[(FragmentId, FragmentId)]) -> bool {
    // Kahn's algorithm: if processed < nodes.len(), a cycle exists
    let mut indeg: HashMap<_, usize> = nodes.iter().map(|n| (*n, 0)).collect();
    let mut adj: HashMap<_, Vec<_>> = HashMap::new();
    for (u, v) in edges { *indeg.entry(*v).or_default() += 1; adj.entry(*u).or_default().push(*v); }
    let mut q: Vec<_> = nodes.iter().filter(|n| indeg[n] == 0).map(|n| *n).collect();
    let mut done = 0;
    while let Some(u) = q.pop() { done += 1; for v in adj.get(&u).cloned().unwrap_or_default() { let d = indeg.get_mut(&v).unwrap(); *d -= 1; if *d == 0 { q.push(v); } } }
    done == nodes.len()
}

When it happens

Trigger: Building a stream graph whose fragment dependency graph contains a cycle — e.g. fragments referencing each other circularly after building edges — so Kahn's algorithm finishes with unprocessed fragments remaining in `downstream_cnts`.

Common situations: Internal meta bugs in edge construction creating circular dependencies between fragments; corrupted job/fragment metadata from a failed migration or manual edits to the catalog.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• table_fragment does not exist: id={0}
• downstream relation missing for {} -> {}
• snapshot backfill epoch set again: {} {} {}
• the upstream fragment should be a MView or Source, got fragm
• the streaming job shouldn't have an upstream fragment, job_t

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/7d60ffe796d210c5. Report an issue: GitHub.