risingwavelabs/risingwave · error

Iceberg source should not have input executor!

Error message

Iceberg source should not have input executor!

What it means

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.

Source

Thrown at src/batch/executors/src/executor/iceberg_scan.rs:159

                        return Ok(());
                    }
                } else {
                    yield chunk;
                }
            }
        }
    }
}

pub struct IcebergScanExecutorBuilder {}

impl BoxedExecutorBuilder for IcebergScanExecutorBuilder {
    async fn new_boxed_executor(
        source: &ExecutorBuilder<'_>,
        inputs: Vec<BoxedExecutor>,
    ) -> crate::error::Result<BoxedExecutor> {
        ensure!(
            inputs.is_empty(),
            "Iceberg source should not have input executor!"
        );
        let source_node = try_match_expand!(
            source.plan_node().get_node_body().unwrap(),
            NodeBody::IcebergScan
        )?;

        // prepare connector source
        let options_with_secret = WithOptionsSecResolved::new(
            source_node.with_properties.clone(),
            source_node.secret_refs.clone(),
        );
        let config = ConnectorProperties::extract(options_with_secret, false)?;

        let split_list = source_node
            .split
            .iter()
            .map(|split| SplitImpl::restore_from_bytes(split).unwrap())

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the barrier manager / recovery path for barriers whose is_checkpoint flag changed between the aligned and incoming stream.
2. After a restart, ensure barrier replay reproduces the exact same barrier sequence (same checkpoint positions).
3. Verify no mixed-version writers/readers are attached to the same log store.
4. Log both barriers (epochs and flags) to identify which stream diverged before fixing the source.

// rust
// compare barrier flags before feeding next_op during alignment
if let Some(aligned) = reader.current_aligned_barrier() {
    anyhow::ensure!(aligned.is_checkpoint == incoming.is_checkpoint,
        "barrier flag mismatch at epoch {} vs {}", aligned.epoch, incoming.epoch);
}

// rust
match reader.next_op().await {
    Ok(op) => handle(op),
    Err(e) if e.to_string().contains("is_checkpoint") => {
        tracing::error!("barrier stream diverged between aligned and incoming: {e:#}");
        // treat as fatal: restart actor / rebuild reader state
    }
    Err(e) => return Err(e),
}

When it happens

Trigger: 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.

Common situations: 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.

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

• Exchange executor should not have children!
• Iceberg metadata scan received a non-Iceberg connector
• missing lagging barriers for direct log-store start from sna
• unable to send barrier
• cannot get truncated epoch

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