risingwavelabs/risingwave · error

Input stream terminated unexpectedly during normal ingestion

Error message

Input stream terminated unexpectedly during normal ingestion

What it means

The materialize executor keeps buffering upstream data in `MaterializeStreamState::Buffering` while waiting for a barrier that lets it switch to normal ingestion. If, while in normal ingestion, the upstream channel yields `None` (stream end) instead of a barrier/chunk, the executor treats it as a fatal topology violation and throws this uncategorized error, because the input to a materialized-view executor must never end during normal operation.

Source

Thrown at src/stream/src/executor/mview/materialize.rs:619

                                        yield Message::Chunk(chunk);
                                    }
                                }
                            }
                            Message::Barrier(barrier) => {
                                *inner_state = MaterializeStreamState::CommitAndYieldBarrier {
                                    barrier,
                                    expect_next_state: Box::new(
                                        MaterializeStreamState::NormalIngestion,
                                    ),
                                };
                                continue 'main_loop;
                            }
                        }
                    }

                    return Err(StreamExecutorError::from(ErrorKind::Uncategorized(
                        anyhow::anyhow!(
                            "Input stream terminated unexpectedly during normal ingestion"
                        ),
                    )));
                }
                MaterializeStreamState::MergingData => {
                    let Some(refresh_args) = self.refresh_args.as_mut() else {
                        panic!(
                            "MaterializeExecutor entered CleanUp state without refresh_args configured"
                        );
                    };
                    tracing::info!(table_id = %refresh_args.table_id, "on_load_finish: Starting table replacement operation");

                    debug_assert_eq!(
                        self.state_table.vnodes(),
                        refresh_args.staging_table.vnodes()
                    );
                    debug_assert_eq!(
                        refresh_args.staging_table.vnodes(),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the upstream actor logs for a crash, panic, or explicit cancellation and fix the root cause (often OOM or a meta-service failover bug)
2. Retry the MV creation or refresh the materialized view if the termination came from a transient cluster event
3. Check whether a concurrent DDL statement (DROP/ALTER) raced with the job; serialize DDL against running MV creation
4. If reproducible without external cause, file an issue — a healthy pipeline must terminate executors via barriers, not channel closure

// Before relying on a long-lived MV, ensure no DDL targets it concurrently
assert_no_concurrent_ddl(materialized_view_name)?;

match ingest_result {
    Err(e) if e.to_string().contains("Input stream terminated unexpectedly") => {
        // Upstream actor died: check upstream logs, then retry after root cause is cleared
        inspect_upstream_actor_logs();
        retry_with_backoff(create_materialized_view);
    }
    other => other?,
}

When it happens

Trigger: The upstream actor/fragment terminates (crashes, is cancelled by a DDL like DROP, or the meta node fails over the fragment) while the materialize executor is in normal ingestion, so the channel closes rather than delivering a `Stop`/barrier message.

Common situations: Concurrent DROP MATERIALIZED VIEW / schema change while the MV is actively ingesting; upstream actor panic or OOM kill during mv creation backfill completion; meta failover bugs that tear down fragments without sending proper barriers.

Related errors

• next offset {:?} should be later than current offset {:?}
• new item epoch {} does not match current chunk offset epoch
• new item epoch {} does not exceed barrier offset epoch {}
• Division by zero
• Array error: {0}

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