{"record":{"id":"fc52fb36d56782cb","repo":"risingwavelabs/risingwave","slug":"0-fc52fb","errorCode":null,"errorMessage":"{0}","messagePattern":"\\{0\\}","errorType":"exception","errorClass":"ProtobufTypeError","httpStatus":null,"severity":"error","filePath":"src/connector/codec/src/decoder/protobuf/parser.rs","lineNumber":53,"sourceCode":"\npub fn pb_schema_to_fields(\n    message_descriptor: &MessageDescriptor,\n    messages_as_jsonb: &HashSet<String>,\n) -> anyhow::Result<Vec<Field>> {\n    let mut parse_trace: Vec<String> = vec![];\n    message_descriptor\n        .fields()\n        .map(|field| {\n            let field_type = protobuf_type_mapping(&field, &mut parse_trace, messages_as_jsonb)\n                .context(\"failed to map protobuf type\")?;\n            let column = Field::new(field.name(), field_type);\n            Ok(column)\n        })\n        .collect()\n}\n\n#[derive(Error, Debug, Macro)]\n#[error(\"{0}\")]\nstruct ProtobufTypeError(#[message] String);\n\nfn detect_loop_and_push(\n    trace: &mut Vec<String>,\n    fd: &FieldDescriptor,\n) -> std::result::Result<(), ProtobufTypeError> {\n    let identifier = format!(\"{}({})\", fd.name(), fd.full_name());\n    if trace.iter().any(|s| s == identifier.as_str()) {\n        bail_protobuf_type_error!(\n            \"circular reference detected: {}, conflict with {}, kind {:?}. Adding {:?} to {:?} may help.\",\n            trace.iter().format(\"->\"),\n            identifier,\n            fd.kind(),\n            fd.kind(),\n            PROTOBUF_MESSAGES_AS_JSONB,\n        );\n    }\n    trace.push(identifier);","sourceCodeStart":35,"sourceCodeEnd":71,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/codec/src/decoder/protobuf/parser.rs#L35-L71","documentation":"ProtobufTypeError is an internal error in the Protobuf parser that reports a type-related problem (as the message string) when deriving RisingWave columns from Protobuf field descriptors. It is thrown during schema mapping, including cycle detection over nested/message-typed fields, when a descriptor's type cannot be converted to a supported RW type.","triggerScenarios":"Parsing a .pb descriptor for a protobuf-encoded source: unsupported field types (e.g. map with unsupported value, oneof edge cases, recursive message types detected by detect_loop_and_push), or unknown well-known types.","commonSituations":"Using a proto with recursive message definitions; protos compiled with protoc versions producing descriptors RW does not handle; using unsupported protobuf types (e.g. google.protobuf.Any) in sources.","solutions":["Remove or restructure recursive/self-referential message types in the proto","Replace unsupported protobuf types (Any, unsupported maps/wrappers) with supported ones","Regenerate the descriptor file with a compatible protoc version matching your RW release","Check the message text for the exact field and consult RW docs on supported proto types"],"exampleFix":"// before: message Node { Node child = 1; } // recursion\n// after\n// message Node { string value = 1; int32 depth = 2; }","handlingStrategy":"validation","validationCode":"// Reject recursive or unsupported messages before creating the source\nfn check(proto: &FileDescriptorSet) -> bool {\n    proto.file.iter().all(|f| f.message_type.iter().all(|m| !is_self_referential(m)))\n}","typeGuard":null,"tryCatchPattern":"match result {\n    Err(ProtobufTypeError(msg)) => {\n        eprintln!(\"protobuf schema unsupported: {msg}\");\n        // fall back to a corrected descriptor or abort source creation\n    }\n    other => other?,\n}","preventionTips":["Avoid recursive message definitions in protos used as sources","Stick to well-supported protobuf types; avoid Any and exotic wrappers","Regenerate descriptors with a protoc version compatible with your RW release"],"tags":["protobuf","schema","parser"],"backgroundTag":"protobuf-unmarshal-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"}