public static SeaTunnelRowType convert(RowType rowType, int[] projectionIndex) {
        String[] fieldNames = rowType.getFieldNames().toArray(new String[0]);
        SeaTunnelDataType<?>[] dataTypes =
                rowType.getFields().stream()
                        .map(field -> field.type().accept(PaimonToSeaTunnelTypeVisitor.INSTANCE))
                        .toArray(SeaTunnelDataType<?>[]::new);
        if (projectionIndex != null) {
            String[] projectionFieldNames =
                    Arrays.stream(projectionIndex)
                            .filter(index -> index >= 0 && index < fieldNames.length)
                            .mapToObj(index -> fieldNames[index])
                            .toArray(String[]::new);
            SeaTunnelDataType<?>[] projectionDataTypes =
                    Arrays.stream(projectionIndex)
                            .filter(index -> index >= 0 && index < fieldNames.length)
                            .mapToObj(index -> dataTypes[index])
                            .toArray(SeaTunnelDataType<?>[]::new);
            return new SeaTunnelRowType(projectionFieldNames, projectionDataTypes);
        }
        return new SeaTunnelRowType(fieldNames, dataTypes);
    }
@Test
    public void paimonToSeaTunnelWithProjection() {
        int[] projection = {7, 2};
        SeaTunnelRowType convert = RowTypeConverter.convert(rowType, projection);
        Assertions.assertEquals(convert, seaTunnelProjectionRowType);
    }