risingwavelabs/risingwave · error

op field not found in canal json

Error message

op field not found in canal json

What it means

After the DDL check, parse_inner reads the `op` field of the Canal JSON event. If it is missing or not one of INSERT/UPDATE/DELETE (Canal values INSERT, UPDATE, DELETE), the parser cannot map the event to a change operation and bails.

Source

Thrown at src/connector/src/parser/canal/simd_json_parser.rs:77

        mut payload: Vec<u8>,
        mut writer: SourceStreamChunkRowWriter<'_>,
    ) -> ConnectorResult<()> {
        let mut event: BorrowedValue<'_> =
            simd_json::to_borrowed_value(&mut payload[self.payload_start_idx..])
                .context("failed to parse canal json payload")?;

        let is_ddl = event
            .get(IS_DDL)
            .and_then(|v| v.as_bool())
            .context("field `isDdl` not found in canal json")?;
        if is_ddl {
            bail!("received a DDL message, please set `canal.instance.filter.query.dml` to true.");
        }

        let op = match event.get(OP).and_then(|v| v.as_str()) {
            Some(CANAL_INSERT_EVENT | CANAL_UPDATE_EVENT) => ChangeEventOperation::Upsert,
            Some(CANAL_DELETE_EVENT) => ChangeEventOperation::Delete,
            _ => bail!("op field not found in canal json"),
        };

        let events = event
            .get_mut(DATA)
            .and_then(|v| match v {
                BorrowedValue::Array(array) => Some(array),
                _ => None,
            })
            .context("field `data` is missing for creating event")?;

        let mut errors = Vec::new();
        for event in events.drain(..) {
            let accessor = JsonAccess::new_with_options(event, &JsonParseOptions::CANAL);
            match apply_row_operation_on_stream_chunk_writer((op, accessor), &mut writer) {
                Ok(_) => {}
                Err(err) => errors.push(err),
            }
        }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure the topic contains only Canal JSON DML data messages; enable heartbeat filtering (`canal.instance.filter.query.dml=true` and disable heartbeat/tx events in MQ routing)
2. Verify the message format is Canal JSON (has isDdl/op/data fields) by inspecting a raw Kafka message
3. Use a compatible source format (e.g. debezium json) if the upstream is not actually Canal

if (!msg || typeof msg.op !== 'string' || !['INSERT','UPDATE','DELETE'].includes(msg.op)) { drop(msg); }

const isCanalDml = (m) => m && typeof m.op === 'string' && ['INSERT','UPDATE','DELETE'].includes(m.op);

if err.to_string().contains("op field not found") { inspect_raw_kafka_message(topic, offset); } else { propagate }

When it happens

Trigger: Message payload lacks an `op` field, or `op` has an unexpected value (e.g. `CREATE`, `QUERY`, `TRUNCATE`) — anything not INSERT/UPDATE/DELETE.

Common situations: Non-Canal-format JSON pushed to the topic; Canal heartbeat or transaction-begin/end events (op=BEGIN/COMMIT) reaching the source; producer format drift or schema registry/message-format changes.

Related errors

• received a DDL message, please set `canal.instance.filter.qu
• failed to parse {} row(s) in a single canal json message: {}
• invalid backfill state: cdc_offset_low
• invalid backfill state: cdc_offset_high
• Field '{}' not found in secret

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