{"record":{"id":"6496a8d100a35202","repo":"risingwavelabs/risingwave","slug":"current-iceberg-schema-does-not-match-either-origi","errorCode":null,"errorMessage":"Current iceberg schema does not match either original_schema ({} cols) or changed schema; cannot determine whether schema change is applied","messagePattern":"Current iceberg schema does not match either original_schema \\((.+?) cols\\) or changed schema; cannot determine whether schema change is applied","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":919,"sourceCode":"                .collect_vec(),\n            _ => {\n                return Err(SinkError::Iceberg(anyhow!(\n                    \"Unsupported sink schema change op in iceberg sink: {:?}\",\n                    schema_change.op\n                )));\n            }\n        };\n\n        // If current schema equals the changed schema, then schema change is applied.\n        if schema_matches(&expected_after_change) {\n            tracing::debug!(\n                \"Current iceberg schema matches changed schema ({} columns); schema change already applied\",\n                expected_after_change.len()\n            );\n            return Ok(true);\n        }\n\n        Err(SinkError::Iceberg(anyhow!(\n            \"Current iceberg schema does not match either original_schema ({} cols) or changed schema; cannot determine whether schema change is applied\",\n            schema_change.original_schema.len()\n        )))\n    }\n\n    /// Commit schema changes (e.g., add columns) to the iceberg table.\n    /// This function uses Transaction API to atomically update the table schema\n    /// with optimistic locking to prevent concurrent conflicts.\n    async fn commit_schema_change_impl(&mut self, schema_change: PbSinkSchemaChange) -> Result<()> {\n        // Step 1: Build new fields to add\n        let iceberg_create_table_arrow_convert = IcebergCreateTableArrowConvert::default();\n        let mut new_fields = Vec::new();\n\n        let mut drop_column_names = Vec::new();\n        match schema_change.op.as_ref() {\n            Some(risingwave_pb::stream_plan::sink_schema_change::Op::AddColumns(\n                add_columns_op,\n            )) => {","sourceCodeStart":901,"sourceCodeEnd":937,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L901-L937","documentation":"check_schema_change_applied compares the current Iceberg table schema against both the pre-change (original_schema) and post-change (expected) column lists to decide if a schema change was already applied. If the current schema matches neither, the function cannot classify the state and fails hard instead of guessing. This prevents wrongly skipping or re-applying a DDL change.","triggerScenarios":"The Iceberg table schema was modified by a third party in a way that diverges from both recorded schemas (extra/renamed/removed columns); the sink's recorded original_schema is stale relative to the actual table; concurrent schema changes interleaved with this check.","commonSituations":"Manual ALTER TABLE on the Iceberg table by another engine; sink rebuilt with outdated schema history; multiple schema changes applied out of order across sink restarts.","solutions":["Inspect the Iceberg table's current schema and reconcile it with the intended original/changed schemas (apply or revert the manual change).","Recreate or refresh the sink so its recorded original_schema matches reality, then re-run the schema change.","Avoid external schema edits on Iceberg tables managed by RisingWave sinks; route all schema changes through the upstream table.","Enable/verify schema evolution support and apply changes in order."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Compare current table schema columns against both expected schemas before relying on auto-detection:\nlet cols: Vec<String> = table.metadata().current_schema().fields().iter().map(|f| f.name.clone()).collect();\nassert!(cols == original_cols || cols == changed_cols, \"iceberg schema diverged from sink's recorded schemas\");","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Don't manually ALTER the Iceberg table outside RisingWave.","Apply schema changes in order and one at a time.","Refresh/recreate the sink after any out-of-band schema edit."],"tags":["iceberg","schema-evolution","state-conflict","sink"],"backgroundTag":"schema-validation-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}