@Override
    public KvMetadata resolveMetadata(
            boolean isKey,
            List<MappingField> resolvedFields,
            Map<String, String> options,
            InternalSerializationService serializationService
    ) {
        Map<QueryPath, MappingField> fieldsByPath = extractFields(resolvedFields, isKey);

        List<TableField> fields = new ArrayList<>();
        for (Entry<QueryPath, MappingField> entry : fieldsByPath.entrySet()) {
            QueryPath path = entry.getKey();
            QueryDataType type = entry.getValue().type();
            String name = entry.getValue().name();

            fields.add(new MapTableField(name, type, false, path));
        }
        maybeAddDefaultField(isKey, resolvedFields, fields, QueryDataType.OBJECT);

        Schema schema = getSchema(fieldsByPath, options, isKey);
        if (schema == null) {
            String recordName = options.getOrDefault(
                    isKey ? OPTION_KEY_AVRO_RECORD_NAME : OPTION_VALUE_AVRO_RECORD_NAME, "jet.sql");
            schema = resolveSchema(recordName, getFields(fieldsByPath));
        }
        return new KvMetadata(
                fields,
                AvroQueryTargetDescriptor.INSTANCE,
                new AvroUpsertTargetDescriptor(schema)
        );
    }
@Test
    public void when_noKeyOrThisPrefixInExternalName_then_usesValue() {
        KvMetadata metadata = INSTANCE.resolveMetadata(
                isKey,
                List.of(field("field", QueryDataType.INT, "extField")),
                emptyMap(),
                null
        );
        assertThat(metadata.getFields()).containsExactly(isKey
                ? new MapTableField[]{
                        new MapTableField("__key", QueryDataType.OBJECT, true, QueryPath.KEY_PATH)
                }
                : new MapTableField[]{
                        new MapTableField("field", QueryDataType.INT, false, new QueryPath("extField", false)),
                        new MapTableField("this", QueryDataType.OBJECT, true, QueryPath.VALUE_PATH)
                });
    }