@VisibleForTesting
    @NotNull protected static Schema toIcebergSchema(
            SeaTunnelRowType rowType, ReadonlyConfig readonlyConfig) {
        Types.StructType structType = SchemaUtils.toIcebergType(rowType).asStructType();
        Set<Integer> identifierFieldIds = new HashSet<>();
        if (Objects.nonNull(readonlyConfig)) {
            List<String> pks =
                    SinkConfig.stringToList(readonlyConfig.get(SinkConfig.TABLE_PRIMARY_KEYS), ",");
            if (CollectionUtils.isNotEmpty(pks)) {
                for (String pk : pks) {
                    Optional<Integer> pkId =
                            structType.fields().stream()
                                    .filter(nestedField -> nestedField.name().equals(pk))
                                    .map(nestedField -> nestedField.fieldId())
                                    .findFirst();
                    if (!pkId.isPresent()) {
                        throw new IllegalArgumentException(
                                String.format(
                                        "iceberg table pk:%s not present in the incoming struct",
                                        pk));
                    }
                    identifierFieldIds.add(pkId.get());
                }
            }
        }
        List<Types.NestedField> fields = new ArrayList<>();
        structType
                .fields()
                .forEach(
                        field -> {
                            fields.add(
                                    identifierFieldIds.contains(field.fieldId())
                                            ? field.asRequired()
                                            : field.asOptional());
                        });
        return new Schema(fields, identifierFieldIds);
    }
@Test
    void testToIcebergSchemaWithoutPk() {
        String[] fieldNames = new String[] {"id", "name", "description", "weight"};
        SeaTunnelDataType<?>[] dataTypes =
                new SeaTunnelDataType[] {
                    BasicType.LONG_TYPE,
                    BasicType.STRING_TYPE,
                    BasicType.STRING_TYPE,
                    BasicType.STRING_TYPE
                };
        SeaTunnelRowType rowType = new SeaTunnelRowType(fieldNames, dataTypes);
        ReadonlyConfig readonlyConfig =
                ReadonlyConfig.fromMap(
                        new HashMap<String, Object>() {
                            {
                            }
                        });
        Schema schema = SchemaUtils.toIcebergSchema(rowType, readonlyConfig);
        Assertions.assertNotNull(schema);
        Assertions.assertEquals(fieldNames.length, schema.columns().size());
        for (Types.NestedField column : schema.columns()) {
            Assertions.assertEquals(fieldNames[column.fieldId() - 1], column.name());
            Assertions.assertEquals(Boolean.FALSE, column.isRequired());
        }
    }