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

        String typeName = getCompactTypeName(fieldsByPath, options, isKey);

        List<TableField> fields = new ArrayList<>(fieldsByPath.size());
        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 = resolveSchema(typeName, getFields(fieldsByPath));

        return new KvMetadata(
                fields,
                GenericQueryTargetDescriptor.DEFAULT,
                new CompactUpsertTargetDescriptor(schema)
        );
    }
@Test
    @Parameters({
            "true, __key",
            "false, this"
    })
    public void test_resolveMetadata(boolean key, String prefix) {
        KvMetadata metadata = INSTANCE.resolveMetadata(
                key,
                List.of(
                        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")
                ),
                Map.of(key ? OPTION_KEY_COMPACT_TYPE_NAME : OPTION_VALUE_COMPACT_TYPE_NAME, "test"),
                createSerializationService()
        );

        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);
        SchemaWriter schemaWriter = new SchemaWriter("test");
        schemaWriter.addField(new FieldDescriptor("boolean", FieldKind.NULLABLE_BOOLEAN));
        schemaWriter.addField(new FieldDescriptor("byte", FieldKind.NULLABLE_INT8));
        schemaWriter.addField(new FieldDescriptor("short", FieldKind.NULLABLE_INT16));
        schemaWriter.addField(new FieldDescriptor("int", FieldKind.NULLABLE_INT32));
        schemaWriter.addField(new FieldDescriptor("long", FieldKind.NULLABLE_INT64));
        schemaWriter.addField(new FieldDescriptor("float", FieldKind.NULLABLE_FLOAT32));
        schemaWriter.addField(new FieldDescriptor("double", FieldKind.NULLABLE_FLOAT64));
        schemaWriter.addField(new FieldDescriptor("decimal", FieldKind.DECIMAL));
        schemaWriter.addField(new FieldDescriptor("string", FieldKind.STRING));
        schemaWriter.addField(new FieldDescriptor("time", FieldKind.TIME));
        schemaWriter.addField(new FieldDescriptor("date", FieldKind.DATE));
        schemaWriter.addField(new FieldDescriptor("timestamp", FieldKind.TIMESTAMP));
        schemaWriter.addField(new FieldDescriptor("timestampTz", FieldKind.TIMESTAMP_WITH_TIMEZONE));
        assertEquals(metadata.getUpsertTargetDescriptor(), new CompactUpsertTargetDescriptor(schemaWriter.build()));
    }