@Override
    public Stream<MappingField> resolveAndValidateFields(
            boolean isKey,
            List<MappingField> userFields,
            Map<String, String> options,
            InternalSerializationService serializationService
    ) {
        Map<QueryPath, MappingField> fieldsByPath = extractFields(userFields, isKey);
        for (QueryPath path : fieldsByPath.keySet()) {
            if (path.isTopLevel()) {
                throw QueryException.error("Cannot use the '" + path + "' field with Avro serialization");
            }
        }

        Schema schema = getSchema(fieldsByPath, options, isKey);
        if (schema != null && options.containsKey("schema.registry.url")) {
            throw new IllegalArgumentException("Inline schema cannot be used with schema registry");
        }

        if (userFields.isEmpty()) {
            if (schema == null) {
                throw QueryException.error(
                        "Either a column list or an inline schema is required to create Avro-based mapping");
            }
            return resolveFields(schema, (name, type) ->
                    new MappingField(name, type, new QueryPath(name, isKey).toString()));
        } else {
            if (schema != null) {
                validate(schema, getFields(fieldsByPath).collect(toList()));
            }
            return fieldsByPath.values().stream();
        }
    }
@Test
    public void when_schemaHasUnsupportedType_then_fieldResolutionFails() {
        List<Schema> unsupportedSchemaTypes = List.of(
                Schema.create(Schema.Type.BYTES),
                SchemaBuilder.array().items(Schema.create(Schema.Type.INT)),
                SchemaBuilder.map().values(Schema.create(Schema.Type.INT)),
                SchemaBuilder.enumeration("enum").symbols("symbol"),
                SchemaBuilder.fixed("fixed").size(0)
        );
        for (Schema schema : unsupportedSchemaTypes) {
            assertThatThrownBy(() -> INSTANCE.resolveAndValidateFields(
                    isKey,
                    emptyList(),
                    Map.of(isKey ? OPTION_KEY_AVRO_SCHEMA : OPTION_VALUE_AVRO_SCHEMA,
                            SchemaBuilder.record("jet.sql").fields()
                                    .name("field").type(schema).noDefault()
                                    .endRecord().toString()),
                    null
            ).toArray()).hasMessage("Unsupported schema type: " + schema.getType());
        }
    }