risingwavelabs/risingwave · error · anyhow::Error

got different schema change

Error message

got different schema change {:?} to prev schema change {:?}

What it means

When collecting a batch of commit requests for an epoch, all handles must report the same schema change (or none). If the first request's schema change differs from a later one, the coordinator cannot produce a single consistent commit and errors out. This guards atomic schema evolution across sink writers.

Solutions

1. Retry the failed epoch after all writers have restarted with the new schema; the coordinator aborts the commit safely.
2. Quiesce the sink (pause it) before applying schema changes, then resume so all writers start with the same schema.
3. Ensure schema-change propagation is atomic — all actors of a fragment should switch schema in the same barrier.
4. Check whether a parallelism alter mixed writers from old and new plan versions and force a full sink restart.

// before resuming a sink after schema change, verify all writers report the same schema version
assert_eq!(writers.iter().map(|w| w.schema_version).collect::<HashSet<_>>().len(), 1);

fn schemas_consistent(changes: &[Option<SchemaChange>]) -> bool {
    changes.iter().all(|c| c == &changes[0])
}

// on this error, restart the sink so all writers reload the same schema, then retry the epoch
if err.contains("got different schema change") { restart_sink_with_new_schema(); }

When it happens

Trigger: Different handles in the same commit batch observe different versions of the table schema — e.g. some writers started before a schema change and some after, or a source schema change lands mid-epoch so only part of the writers pick it up.

Common situations: Running `ALTER`/schema-change operations on the sink's upstream table while the sink is actively committing; skewed actor deployments after a parallelism change; stale writers replaying old schema metadata.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• empty sink metadata
• end of writer request stream
• expect AlignInitialEpoch but got
• expect new handle during init, but got
• failed to ack aligned initial epoch

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

Appendix: source

Thrown at src/meta/src/manager/sink_coordination/coordinator_worker.rs:823

            pending_epochs.entry(epoch).or_default().add_new_request(
                handle_id,
                commit_request,
                self.handle_manager.vnode_bitmap(handle_id),
            )?;
            if pending_epochs
                .first_key_value()
                .expect("non-empty")
                .1
                .aligned()
            {
                let (epoch, commit_requests) = pending_epochs.pop_first().expect("non-empty");
                let mut metadatas = Vec::with_capacity(commit_requests.requests.len());
                let mut requests = commit_requests.requests.into_iter();
                let (first_metadata, first_schema_change) = requests.next().expect("non-empty");
                metadatas.push(first_metadata);
                for (metadata, schema_change) in requests {
                    if first_schema_change != schema_change {
                        return Err(anyhow!(
                            "got different schema change {:?} to prev schema change {:?}",
                            schema_change,
                            first_schema_change
                        ));
                    }
                    metadatas.push(metadata);
                }

                match &mut coordinator {
                    SinkCommitCoordinator::SinglePhase(coordinator) => {
                        if !metadatas.is_empty() {
                            let start_time = Instant::now();
                            run_future_with_periodic_fn(
                                coordinator.commit_data(epoch, metadatas).instrument_await(
                                    Self::commit_span("single_phase_commit_data", sink_id, epoch),
                                ),
                                Duration::from_secs(5),
                                || {

View on GitHub (pinned to 6469eb736d)