public static ResolvedSchema expandCompositeTypeToSchema(DataType dataType) {
        if (dataType instanceof FieldsDataType) {
            return expandCompositeType((FieldsDataType) dataType);
        } else if (dataType.getLogicalType() instanceof LegacyTypeInformationType
                && dataType.getLogicalType().getTypeRoot() == STRUCTURED_TYPE) {
            return expandLegacyCompositeType(dataType);
        }

        throw new IllegalArgumentException("Expected a composite type");
    }
@Test
    void testExpandStructuredType() {
        StructuredType logicalType =
                StructuredType.newBuilder(ObjectIdentifier.of("catalog", "database", "type"))
                        .attributes(
                                Arrays.asList(
                                        new StructuredType.StructuredAttribute(
                                                "f0", DataTypes.INT().getLogicalType()),
                                        new StructuredType.StructuredAttribute(
                                                "f1", DataTypes.STRING().getLogicalType()),
                                        new StructuredType.StructuredAttribute(
                                                "f2", DataTypes.TIMESTAMP(5).getLogicalType()),
                                        new StructuredType.StructuredAttribute(
                                                "f3", DataTypes.TIMESTAMP(3).getLogicalType())))
                        .build();

        List<DataType> dataTypes =
                Arrays.asList(
                        DataTypes.INT(),
                        DataTypes.STRING(),
                        DataTypes.TIMESTAMP(5).bridgedTo(Timestamp.class),
                        DataTypes.TIMESTAMP(3));
        FieldsDataType dataType = new FieldsDataType(logicalType, dataTypes);

        ResolvedSchema schema = DataTypeUtils.expandCompositeTypeToSchema(dataType);

        assertThat(schema)
                .isEqualTo(
                        ResolvedSchema.of(
                                Column.physical("f0", INT()),
                                Column.physical("f1", STRING()),
                                Column.physical("f2", TIMESTAMP(5).bridgedTo(Timestamp.class)),
                                Column.physical(
                                        "f3", TIMESTAMP(3).bridgedTo(LocalDateTime.class))));
    }