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

        PortableId portableId = getPortableId(fieldsByPath, options, isKey);
        ClassDefinition classDefinition = resolveClassDefinition(portableId, getFields(fieldsByPath),
                serializationService.getPortableContext());

        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);

        return new KvMetadata(
                fields,
                GenericQueryTargetDescriptor.DEFAULT,
                new PortableUpsertTargetDescriptor(classDefinition)
        );
    }
@Test
    @Parameters({
            "true, __key",
            "false, this"
    })
    public void test_resolveMetadata(boolean key, String prefix) {
        KvMetadata metadata = INSTANCE.resolveMetadata(
                key,
                asList(
                        field("boolean", QueryDataType.BOOLEAN, prefix + ".boolean"),
                        field("byte", QueryDataType.TINYINT, prefix + ".byte"),
                        field("short", QueryDataType.SMALLINT, prefix + ".short"),
                        field("int", QueryDataType.INT, prefix + ".int"),
                        field("long", QueryDataType.BIGINT, prefix + ".long"),
                        field("float", QueryDataType.REAL, prefix + ".float"),
                        field("double", QueryDataType.DOUBLE, prefix + ".double"),
                        field("decimal", QueryDataType.DECIMAL, prefix + ".decimal"),
                        field("string", QueryDataType.VARCHAR, prefix + ".string"),
                        field("time", QueryDataType.TIME, prefix + ".time"),
                        field("date", QueryDataType.DATE, prefix + ".date"),
                        field("timestamp", QueryDataType.TIMESTAMP, prefix + ".timestamp"),
                        field("timestampTz", QueryDataType.TIMESTAMP_WITH_TZ_OFFSET_DATE_TIME, prefix + ".timestampTz")
                ),
                ImmutableMap.of(
                        (key ? OPTION_KEY_FACTORY_ID : OPTION_VALUE_FACTORY_ID), "1",
                        (key ? OPTION_KEY_CLASS_ID : OPTION_VALUE_CLASS_ID), "2",
                        (key ? OPTION_KEY_CLASS_VERSION : OPTION_VALUE_CLASS_VERSION), "3"
                ),
                new DefaultSerializationServiceBuilder().build()
        );

        assertThat(metadata.getFields()).containsExactly(
                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("string", QueryDataType.VARCHAR, false, QueryPath.create(prefix + ".string")),
                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(prefix, QueryDataType.OBJECT, true, QueryPath.create(prefix))
        );
        assertThat(metadata.getQueryTargetDescriptor()).isEqualTo(GenericQueryTargetDescriptor.DEFAULT);
        assertThat(metadata.getUpsertTargetDescriptor())
                .isEqualToComparingFieldByField(new PortableUpsertTargetDescriptor(
                        new ClassDefinitionBuilder(1, 2, 3)
                                .addBooleanField("boolean")
                                .addByteField("byte")
                                .addShortField("short")
                                .addIntField("int")
                                .addLongField("long")
                                .addFloatField("float")
                                .addDoubleField("double")
                                .addDecimalField("decimal")
                                .addStringField("string")
                                .addTimeField("time")
                                .addDateField("date")
                                .addTimestampField("timestamp")
                                .addTimestampWithTimezoneField("timestampTz")
                                .build()
                ));
    }