{"record":{"id":"9af62f29df45e35d","repo":"risingwavelabs/risingwave","slug":"schema-invalid-record-type-required-at-top-level","errorCode":null,"errorMessage":"schema invalid, record type required at top level of the schema.","messagePattern":"schema invalid, record type required at top level of the schema\\.","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/codec/src/decoder/avro/schema.rs","lineNumber":88,"sourceCode":"    }\n}\n\n/// This function expects original schema (with `Ref`).\n/// TODO: change `map_handling` to some `Config`, and also unify debezium.\npub fn avro_schema_to_fields(\n    schema: &Schema,\n    map_handling: Option<MapHandling>,\n) -> anyhow::Result<Vec<Field>> {\n    let resolved = ResolvedSchema::try_from(schema)?;\n    let mut ancestor_records: Vec<String> = vec![];\n    let root_type = avro_type_mapping(\n        schema,\n        &mut ancestor_records,\n        resolved.get_names(),\n        map_handling,\n    )?;\n    let DataType::Struct(root_struct) = root_type else {\n        bail!(\"schema invalid, record type required at top level of the schema.\");\n    };\n    let fields = root_struct\n        .iter()\n        .map(|(name, data_type)| Field::new(name, data_type.clone()))\n        .collect();\n    Ok(fields)\n}\n\nconst DBZ_VARIABLE_SCALE_DECIMAL_NAME: &str = \"VariableScaleDecimal\";\nconst DBZ_VARIABLE_SCALE_DECIMAL_NAMESPACE: &str = \"io.debezium.data\";\n\n/// This function expects original schema (with `Ref`).\nfn avro_type_mapping(\n    schema: &Schema,\n    ancestor_records: &mut Vec<String>,\n    refs: &NamesRef<'_>,\n    map_handling: Option<MapHandling>,\n) -> anyhow::Result<DataType> {","sourceCodeStart":70,"sourceCodeEnd":106,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/codec/src/decoder/avro/schema.rs#L70-L106","documentation":"avro_schema_to_fields requires the resolved root of the Avro schema to map to a RisingWave Struct type, which corresponds to an Avro record. If the top-level type is anything else (e.g. array, primitive, map), the root cannot be turned into a list of fields and the function bails. This enforces that Avro schemas describe a top-level record.","triggerScenarios":"Calling json_schema_to_columns, avro_schema_str_to_risingwave_schema, map_to_columns, or extract_pks with a JSON/Avro schema whose top-level type is not \"record\" (e.g. {\"type\":\"array\"} or a bare primitive schema).","commonSituations":"Loading a schema registry entry that is an array-of-records instead of a record; hand-writing a minimal Avro schema with a primitive top type; concatenating or trimming schema JSON incorrectly.","solutions":["Wrap the schema in a top-level record type with named fields.","If the payload is an array of records, extract the record schema or adjust the upstream producer to emit record-rooted schemas.","Verify the fetched schema URL/registry subject actually points to the record schema."],"exampleFix":"// before\n{\"type\": \"array\", \"items\": {\"type\": \"record\", \"name\": \"Event\", \"fields\": [...]}}\n// after\n{\"type\": \"record\", \"name\": \"Envelope\", \"fields\": [{\"name\": \"events\", \"type\": {\"type\": \"array\", \"items\": {\"type\": \"record\", \"name\": \"Event\", \"fields\": [...]}}}]}","handlingStrategy":"validation","validationCode":"const schema = JSON.parse(schemaStr);\nif (schema.type !== \"record\") {\n  throw new Error(`top-level Avro type must be record, got ${schema.type}`);\n}","typeGuard":"function isRecordSchema(s) { return typeof s === 'object' && s !== null && s.type === 'record' && Array.isArray(s.fields); }","tryCatchPattern":null,"preventionTips":["Always wrap payloads in a top-level record type when generating Avro schemas.","Fetch and eyeball the registry schema before wiring it into the source.","Add a schema linting step in CI that asserts top-level record."],"tags":["avro","schema","validation"],"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"}