risingwavelabs/risingwave · error · SinkError::Iceberg

schema_id and partition_spec_id should be the same in all wr

Error message

schema_id and partition_spec_id should be the same in all write results

What it means

pre_commit_inner validates that every write result in the batch used the same schema_id and partition_spec_id expected for the table (expect_schema_id / expect_partition_spec_id). If any result diverges, the snapshot would mix files from incompatible schemas/specs, so it aborts with this error before committing to Iceberg.

Source

Thrown at src/connector/src/sink/iceberg/commit.rs:625

        );

        // Skip if no data to commit
        if write_results.is_empty() || write_results.iter().all(|r| r.data_files.is_empty()) {
            return Ok(None);
        }

        let expect_schema_id = write_results[0].schema_id;
        let expect_partition_spec_id = write_results[0].partition_spec_id;

        // guarantee that all write results has same schema_id and partition_spec_id
        if write_results
            .iter()
            .any(|r| r.schema_id != expect_schema_id)
            || write_results
                .iter()
                .any(|r| r.partition_spec_id != expect_partition_spec_id)
        {
            return Err(SinkError::Iceberg(anyhow!(
                "schema_id and partition_spec_id should be the same in all write results"
            )));
        }

        let snapshot_id = FastAppendAction::generate_snapshot_id(&self.table);
        tracing::debug!(
            iceberg_component = "sink_committer",
            iceberg_operation = "pre_commit",
            sink_id = %self.sink_id,
            table = %self.table.identifier(),
            epoch,
            snapshot_id,
            schema_id = expect_schema_id,
            partition_spec_id = expect_partition_spec_id,
            data_file_count,
            "iceberg_sink_pre_commit_snapshot_assigned",
        );

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure no schema/partition evolution happens while a sink epoch is in flight (pause evolution or drain the sink first)
2. Restart/rebuild the sink so all actors use the current schema_id and partition_spec_id
3. Verify the table's current schema/spec matches what the sink was created with and update the sink definition if it drifted
4. Check for concurrent writers (other engines) evolving the table and coordinate external evolution

fn write_results_consistent(results: &[IcebergCommitResult], schema_id: i32, spec_id: i32) -> bool {
    results.iter().all(|r| r.schema_id == schema_id && r.partition_spec_id == spec_id)
}

if !write_results_consistent(&write_results, expect_schema_id, expect_partition_spec_id) {
    return Err(SinkError::Iceberg(anyhow!("inconsistent schema_id/partition_spec_id in write results")));
}

When it happens

Trigger: A sink epoch contains write results produced under a different table schema or partition spec than the current one — e.g. the table was evolved (ALTER / schema evolution / partition spec rewrite) while in-flight write results from the old schema were still being committed.

Common situations: Concurrent Iceberg table schema/partition evolution by another engine or a RisingWave schema change mid-epoch; sinks recovered across a table definition change; stale actors continuing to write after schema evolution.

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

• iceberg sink: schema evolution not supported; expect schema
• Delete field id {} not found in schema
• table {} not found
• register_table is not supported in mock catalog
• iceberg sink metadata should have schema_id

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