{"record":{"id":"20880d44f6f18c9e","repo":"risingwavelabs/risingwave","slug":"failed-to-parse-row-s-in-a-single-canal-json-m","errorCode":null,"errorMessage":"failed to parse {} row(s) in a single canal json message: {}","messagePattern":"failed to parse (.+?) row\\(s\\) in a single canal json message: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/parser/canal/simd_json_parser.rs","lineNumber":101,"sourceCode":"                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(..) {\n            let accessor = JsonAccess::new_with_options(event, &JsonParseOptions::CANAL);\n            match apply_row_operation_on_stream_chunk_writer((op, accessor), &mut writer) {\n                Ok(_) => {}\n                Err(err) => errors.push(err),\n            }\n        }\n\n        if errors.is_empty() {\n            Ok(())\n        } else {\n            // TODO(error-handling): multiple errors\n            bail!(\n                \"failed to parse {} row(s) in a single canal json message: {}\",\n                errors.len(),\n                errors.iter().format(\", \")\n            );\n        }\n    }\n}\n\nimpl ByteStreamSourceParser for CanalJsonParser {\n    fn columns(&self) -> &[SourceColumnDesc] {\n        &self.rw_columns\n    }\n\n    fn source_ctx(&self) -> &SourceContext {\n        &self.source_ctx\n    }\n\n    fn parser_format(&self) -> ParserFormat {","sourceCodeStart":83,"sourceCodeEnd":119,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/parser/canal/simd_json_parser.rs#L83-L119","documentation":"A single Canal JSON message may carry multiple rows in its `data` array. Rows are parsed individually and errors collected; if any row fails, the whole message fails with this aggregated error listing the count and all underlying row errors.","triggerScenarios":"Any of the row-level parse failures (column mismatch, JSON value type not convertible to the target column type, etc.) in a multi-row `data` array.","commonSituations":"Schema drift between MySQL table and the RW source columns (added/renamed/retyped columns); bad data types inserted upstream (e.g. oversized strings into SMALLINT); messages produced with multiple rows per binlog event.","solutions":["Read the joined inner error messages to identify the offending rows/columns","Align the source schema with the MySQL table (recreate source or alter columns) after schema changes upstream","Fix or sanitize the upstream data causing the per-row failures","If partial ingestion is acceptable, split upstream messages to single-row events so only the bad row fails"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// pre-validate rows against expected columns\nfor (const row of msg.data) { for (const col of requiredColumns) if (!(col in row)) throw new Error(`row missing column ${col}`); }","typeGuard":null,"tryCatchPattern":"match res { Err(e) if e.to_string().starts_with(\"failed to parse\") => { log_full(e); route_to_dlq(msg); } , Err(e) => propagate(e), Ok(v) => v }","preventionTips":["Keep MySQL table schema and source schema in sync; run schema-drift checks","Split large multi-row binlog events upstream to isolate bad rows","Use a DLQ for unparseable messages instead of stalling the source"],"tags":["canal","cdc","json","batch-parse"],"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"}