{"record":{"id":"4abaa1cb796e34b8","repo":"risingwavelabs/risingwave","slug":"iceberg-source-should-not-have-input-executor","errorCode":null,"errorMessage":"Iceberg source should not have input executor!","messagePattern":"Iceberg source should not have input executor!","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/batch/executors/src/executor/iceberg_scan.rs","lineNumber":159,"sourceCode":"                        return Ok(());\n                    }\n                } else {\n                    yield chunk;\n                }\n            }\n        }\n    }\n}\n\npub struct IcebergScanExecutorBuilder {}\n\nimpl BoxedExecutorBuilder for IcebergScanExecutorBuilder {\n    async fn new_boxed_executor(\n        source: &ExecutorBuilder<'_>,\n        inputs: Vec<BoxedExecutor>,\n    ) -> crate::error::Result<BoxedExecutor> {\n        ensure!(\n            inputs.is_empty(),\n            \"Iceberg source should not have input executor!\"\n        );\n        let source_node = try_match_expand!(\n            source.plan_node().get_node_body().unwrap(),\n            NodeBody::IcebergScan\n        )?;\n\n        // prepare connector source\n        let options_with_secret = WithOptionsSecResolved::new(\n            source_node.with_properties.clone(),\n            source_node.secret_refs.clone(),\n        );\n        let config = ConnectorProperties::extract(options_with_secret, false)?;\n\n        let split_list = source_node\n            .split\n            .iter()\n            .map(|split| SplitImpl::restore_from_bytes(split).unwrap())","sourceCodeStart":141,"sourceCodeEnd":177,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/executors/src/executor/iceberg_scan.rs#L141-L177","documentation":"During barrier alignment in the decode stream (`check_is_checkpoint`, used by `next_op`), a barrier was delivered that does not match the barrier currently being aligned: their `is_checkpoint` flags differ. The barrier stream must be identical across replays/readers, so a checkpoint-barrier vs plain-barrier mismatch is treated as a fatal inconsistency.","triggerScenarios":"Calling `next_op` while the `AlignedBarrier` stream state holds a barrier with `is_checkpoint` value X, and the incoming barrier has `is_checkpoint` != X. Thrown at serde.rs:606.","commonSituations":"Mismatched barrier sequences after actor recovery (replayed barriers differ from current ones); changes in barrier checkpoint marking between versions; multiple readers with divergent barrier streams during migration or failover.","solutions":["Check the barrier manager / recovery path for barriers whose is_checkpoint flag changed between the aligned and incoming stream.","After a restart, ensure barrier replay reproduces the exact same barrier sequence (same checkpoint positions).","Verify no mixed-version writers/readers are attached to the same log store.","Log both barriers (epochs and flags) to identify which stream diverged before fixing the source."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// rust\n// compare barrier flags before feeding next_op during alignment\nif let Some(aligned) = reader.current_aligned_barrier() {\n    anyhow::ensure!(aligned.is_checkpoint == incoming.is_checkpoint,\n        \"barrier flag mismatch at epoch {} vs {}\", aligned.epoch, incoming.epoch);\n}","typeGuard":null,"tryCatchPattern":"// rust\nmatch reader.next_op().await {\n    Ok(op) => handle(op),\n    Err(e) if e.to_string().contains(\"is_checkpoint\") => {\n        tracing::error!(\"barrier stream diverged between aligned and incoming: {e:#}\");\n        // treat as fatal: restart actor / rebuild reader state\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Guarantee barrier replay reproduces identical checkpoint positions after recovery.","Avoid mixing versions of writers/readers on one log store.","Log is_checkpoint flags of barriers during recovery to catch divergence early."],"tags":["log-store","barrier","checkpoint","alignment"],"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"}