@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_objectIsForbiddenForCompact(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("object", QueryDataType.OBJECT, prefix + ".object"));

        // TODO: fix compact nested types support?
        assertThatThrownBy(() -> INSTANCE.resolveAndValidateFields(key, fields, options, ss).collect(toList()))
                .isInstanceOf(QueryException.class)
                .hasMessageContaining("Cannot derive Compact type for '" + QueryDataTypeFamily.OBJECT + "'");
    }