What causes "schema invalid, record type required at top level of the schema." in risingwavelabs/risingwave?

Wrap the schema in a top-level record type with named fields.

schema invalid, record type required at top level of the sch

Error message

schema invalid, record type required at top level of the schema.

What it means

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.

Source

Thrown at src/connector/codec/src/decoder/avro/schema.rs:88

    }
}

/// This function expects original schema (with `Ref`).
/// TODO: change `map_handling` to some `Config`, and also unify debezium.
pub fn avro_schema_to_fields(
    schema: &Schema,
    map_handling: Option<MapHandling>,
) -> anyhow::Result<Vec<Field>> {
    let resolved = ResolvedSchema::try_from(schema)?;
    let mut ancestor_records: Vec<String> = vec![];
    let root_type = avro_type_mapping(
        schema,
        &mut ancestor_records,
        resolved.get_names(),
        map_handling,
    )?;
    let DataType::Struct(root_struct) = root_type else {
        bail!("schema invalid, record type required at top level of the schema.");
    };
    let fields = root_struct
        .iter()
        .map(|(name, data_type)| Field::new(name, data_type.clone()))
        .collect();
    Ok(fields)
}

const DBZ_VARIABLE_SCALE_DECIMAL_NAME: &str = "VariableScaleDecimal";
const DBZ_VARIABLE_SCALE_DECIMAL_NAMESPACE: &str = "io.debezium.data";

/// This function expects original schema (with `Ref`).
fn avro_type_mapping(
    schema: &Schema,
    ancestor_records: &mut Vec<String>,
    refs: &NamesRef<'_>,
    map_handling: Option<MapHandling>,
) -> anyhow::Result<DataType> {

View on GitHub (pinned to 6469eb736d)

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.

Example fix

// before
{"type": "array", "items": {"type": "record", "name": "Event", "fields": [...]}}
// after
{"type": "record", "name": "Envelope", "fields": [{"name": "events", "type": {"type": "array", "items": {"type": "record", "name": "Event", "fields": [...]}}}]}

Defensive patterns

Strategy: validation

Validate before calling

const schema = JSON.parse(schemaStr);
if (schema.type !== "record") {
  throw new Error(`top-level Avro type must be record, got ${schema.type}`);
}

Type guard

function isRecordSchema(s) { return typeof s === 'object' && s !== null && s.type === 'record' && Array.isArray(s.fields); }

Prevention

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.

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

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