@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 test_resolveMetadata() {
        KvMetadata metadata = INSTANCE.resolveMetadata(
                isKey,
                List.of(
                        field("string", QueryDataType.VARCHAR),
                        field("boolean", QueryDataType.BOOLEAN),
                        field("byte", QueryDataType.TINYINT),
                        field("short", QueryDataType.SMALLINT),
                        field("int", QueryDataType.INT),
                        field("long", QueryDataType.BIGINT),
                        field("float", QueryDataType.REAL),
                        field("double", QueryDataType.DOUBLE),
                        field("decimal", QueryDataType.DECIMAL),
                        field("time", QueryDataType.TIME),
                        field("date", QueryDataType.DATE),
                        field("timestamp", QueryDataType.TIMESTAMP),
                        field("timestampTz", QueryDataType.TIMESTAMP_WITH_TZ_OFFSET_DATE_TIME),
                        field("object", QueryDataType.OBJECT)
                ),
                emptyMap(),
                null
        );

        assertThat(metadata.getFields()).containsExactly(
                new MapTableField("string", QueryDataType.VARCHAR, false, QueryPath.create(prefix + ".string")),
                new MapTableField("boolean", QueryDataType.BOOLEAN, false, QueryPath.create(prefix + ".boolean")),
                new MapTableField("byte", QueryDataType.TINYINT, false, QueryPath.create(prefix + ".byte")),
                new MapTableField("short", QueryDataType.SMALLINT, false, QueryPath.create(prefix + ".short")),
                new MapTableField("int", QueryDataType.INT, false, QueryPath.create(prefix + ".int")),
                new MapTableField("long", QueryDataType.BIGINT, false, QueryPath.create(prefix + ".long")),
                new MapTableField("float", QueryDataType.REAL, false, QueryPath.create(prefix + ".float")),
                new MapTableField("double", QueryDataType.DOUBLE, false, QueryPath.create(prefix + ".double")),
                new MapTableField("decimal", QueryDataType.DECIMAL, false, QueryPath.create(prefix + ".decimal")),
                new MapTableField("time", QueryDataType.TIME, false, QueryPath.create(prefix + ".time")),
                new MapTableField("date", QueryDataType.DATE, false, QueryPath.create(prefix + ".date")),
                new MapTableField("timestamp", QueryDataType.TIMESTAMP, false, QueryPath.create(prefix + ".timestamp")),
                new MapTableField("timestampTz", QueryDataType.TIMESTAMP_WITH_TZ_OFFSET_DATE_TIME, false,
                        QueryPath.create(prefix + ".timestampTz")),
                new MapTableField("object", QueryDataType.OBJECT, false, QueryPath.create(prefix + ".object")),
                new MapTableField(prefix, QueryDataType.OBJECT, true, QueryPath.create(prefix))
        );
        assertThat(metadata.getQueryTargetDescriptor()).isEqualTo(AvroQueryTargetDescriptor.INSTANCE);
        assertThat(metadata.getUpsertTargetDescriptor()).isEqualToComparingFieldByField(
                new AvroUpsertTargetDescriptor(SchemaBuilder.record("jet.sql").fields()
                        .optionalString("string")
                        .optionalBoolean("boolean")
                        .optionalInt("byte")
                        .optionalInt("short")
                        .optionalInt("int")
                        .optionalLong("long")
                        .optionalFloat("float")
                        .optionalDouble("double")
                        .optionalString("decimal")
                        .optionalString("time")
                        .optionalString("date")
                        .optionalString("timestamp")
                        .optionalString("timestampTz")
                        .name("object").type(OBJECT_SCHEMA).withDefault(null)
                        .endRecord()));
    }