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 testExpandRowType() {
        DataType dataType =
                ROW(
                        FIELD("f0", INT()),
                        FIELD("f1", STRING()),
                        FIELD("f2", TIMESTAMP(5).bridgedTo(Timestamp.class)),
                        FIELD("f3", TIMESTAMP(3)));
        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))));
    }