@Override
    public Stream<MappingField> resolveAndValidateFields(
            boolean isKey,
            List<MappingField> userFields,
            Map<String, String> options,
            InternalSerializationService serializationService
    ) {
        if (userFields.isEmpty()) {
            throw QueryException.error("Column list is required for Compact format");
        }
        Map<QueryPath, MappingField> fieldsByPath = extractFields(userFields, isKey);

        // Check if the compact type name is specified
        getCompactTypeName(fieldsByPath, options, isKey);

        return fieldsByPath.entrySet().stream()
                .map(entry -> {
                    QueryPath path = entry.getKey();
                    if (path.isTopLevel()) {
                        throw QueryException.error("Cannot use the '" + path + "' field with Compact serialization");
                    }
                    QueryDataType type = entry.getValue().type();
                    if (type == QueryDataType.OBJECT) {
                        throw QueryException.error("Cannot derive Compact type for '" + type.getTypeFamily() + "'");
                    }
                    return entry.getValue();
                });
    }
@Test
    @Parameters({
            "true, __key",
            "false, this"
    })
    public void test_resolveFields(boolean key, String prefix) {
        InternalSerializationService ss = createSerializationService();
        Map<String, String> options =
                Map.of(key ? OPTION_KEY_COMPACT_TYPE_NAME : OPTION_VALUE_COMPACT_TYPE_NAME, "testAll");

        List<MappingField> fields = List.of(
                field("string", QueryDataType.VARCHAR, prefix + ".string"),
                field("character", QueryDataType.VARCHAR_CHARACTER, prefix + ".character"),
                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("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")
        );

        Stream<MappingField> resolvedFields = MetadataCompactResolver.INSTANCE.resolveAndValidateFields(key, fields, options, ss);

        assertThat(resolvedFields).containsExactlyElementsOf(fields);
    }