private static ResolvedSchema expandLegacyCompositeType(DataType dataType) {
        // legacy composite type
        CompositeType<?> compositeType =
                (CompositeType<?>)
                        ((LegacyTypeInformationType<?>) dataType.getLogicalType())
                                .getTypeInformation();

        String[] fieldNames = compositeType.getFieldNames();
        DataType[] fieldTypes =
                Arrays.stream(fieldNames)
                        .map(compositeType::getTypeAt)
                        .map(TypeConversions::fromLegacyInfoToDataType)
                        .toArray(DataType[]::new);

        return ResolvedSchema.physical(fieldNames, fieldTypes);
    }
@Test
    void testExpandLegacyCompositeType() {
        DataType dataType =
                TypeConversions.fromLegacyInfoToDataType(
                        new TupleTypeInfo<>(Types.STRING, Types.INT, Types.SQL_TIMESTAMP));
        ResolvedSchema schema = DataTypeUtils.expandCompositeTypeToSchema(dataType);

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