{"record":{"id":"d7ac18cf81b46c30","repo":"risingwavelabs/risingwave","slug":"received-a-ddl-message-please-set-canal-instance","errorCode":null,"errorMessage":"received a DDL message, please set `canal.instance.filter.query.dml` to true.","messagePattern":"received a DDL message, please set `canal\\.instance\\.filter\\.query\\.dml` to true\\.","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/parser/canal/simd_json_parser.rs","lineNumber":71,"sourceCode":"        })\n    }\n\n    #[expect(clippy::unused_async)]\n    pub async fn parse_inner(\n        &self,\n        mut payload: Vec<u8>,\n        mut writer: SourceStreamChunkRowWriter<'_>,\n    ) -> ConnectorResult<()> {\n        let mut event: BorrowedValue<'_> =\n            simd_json::to_borrowed_value(&mut payload[self.payload_start_idx..])\n                .context(\"failed to parse canal json payload\")?;\n\n        let is_ddl = event\n            .get(IS_DDL)\n            .and_then(|v| v.as_bool())\n            .context(\"field `isDdl` not found in canal json\")?;\n        if is_ddl {\n            bail!(\"received a DDL message, please set `canal.instance.filter.query.dml` to true.\");\n        }\n\n        let op = match event.get(OP).and_then(|v| v.as_str()) {\n            Some(CANAL_INSERT_EVENT | CANAL_UPDATE_EVENT) => ChangeEventOperation::Upsert,\n            Some(CANAL_DELETE_EVENT) => ChangeEventOperation::Delete,\n            _ => bail!(\"op field not found in canal json\"),\n        };\n\n        let events = event\n            .get_mut(DATA)\n            .and_then(|v| match v {\n                BorrowedValue::Array(array) => Some(array),\n                _ => None,\n            })\n            .context(\"field `data` is missing for creating event\")?;\n\n        let mut errors = Vec::new();\n        for event in events.drain(..) {","sourceCodeStart":53,"sourceCodeEnd":89,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/parser/canal/simd_json_parser.rs#L53-L89","documentation":"The Canal JSON parser refuses messages whose `isDdl` flag is true. RisingWave consumes only DML change events; DDL statements must be filtered at the Canal server so they never appear on the Kafka topic.","triggerScenarios":"A Canal instance was configured without `canal.instance.filter.query.dml=true`, so DDL events (CREATE/ALTER/DROP) are serialized into the topic and parsed by parse_inner.","commonSituations":"Default Canal deployments that don't filter DDL; adding DDL to a replicated database after the source was set up; sharing a Canal topic between consumers where one needs DDL.","solutions":["Set `canal.instance.filter.query.dml=true` in the Canal instance configuration and restart the instance","Filter the topic on the producer side (e.g. Canal MQ routing rules) to drop DDL messages","Use a separate Canal topic/instance for DDL if DDL capture is needed elsewhere"],"exampleFix":"// canal.properties / instance.properties (before)\n# canal.instance.filter.query.dml unset\n// after\ncanal.instance.filter.query.dml=true","handlingStrategy":"validation","validationCode":"if (msg && msg.isDdl === true) { /* route to DDL handler or drop before feeding RW */ }","typeGuard":"const isDml = (m) => m && m.isDdl === false && typeof m.op === 'string';","tryCatchPattern":"if err.to_string().contains(\"DDL message\") { drop_event_and_metric(\"canal_ddl_filtered\"); } else { propagate }","preventionTips":["Always set canal.instance.filter.query.dml=true for CDC sources feeding RisingWave","Test the Canal instance config with a dummy DDL before production","Monitor the topic for DDL events with a side consumer"],"tags":["canal","cdc","mysql","streaming-parser"],"backgroundTag":"unsupported-operation","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"}