public static Schema toIcebergApiSchema(List<Column> columns) {
        List<Types.NestedField> icebergColumns = new ArrayList<>();
        for (Column column : columns) {
            int index = icebergColumns.size();
            org.apache.iceberg.types.Type type = toIcebergColumnType(column.getType());
            String colComment = StringUtils.defaultIfBlank(column.getComment(), null);
            Types.NestedField field = Types.NestedField.of(
                    index, column.isAllowNull(), column.getName(), type, colComment);
            icebergColumns.add(field);
        }

        org.apache.iceberg.types.Type icebergSchema = Types.StructType.of(icebergColumns);
        AtomicInteger nextFieldId = new AtomicInteger(1);
        icebergSchema = TypeUtil.assignFreshIds(icebergSchema, nextFieldId::getAndIncrement);
        return new Schema(icebergSchema.asStructType().fields());
    }
@Test
    public void testToIcebergApiSchema() {
        List<Column> columns = Lists.newArrayList();
        columns.add(new Column("c1", Type.BOOLEAN));
        columns.add(new Column("c2", Type.INT));
        columns.add(new Column("c3", Type.BIGINT));
        columns.add(new Column("c4", Type.FLOAT));
        columns.add(new Column("c5", Type.DOUBLE));
        columns.add(new Column("c6", Type.DATE));
        columns.add(new Column("c7", Type.DATETIME));
        columns.add(new Column("c8", Type.VARCHAR));
        columns.add(new Column("c9", Type.CHAR));
        columns.add(new Column("c10", Type.DECIMAL32));
        columns.add(new Column("c11", Type.DECIMAL64));
        columns.add(new Column("c12", Type.DECIMAL128));
        columns.add(new Column("c13", new ArrayType(Type.INT)));
        columns.add(new Column("c14", new MapType(Type.INT, Type.INT)));
        columns.add(new Column("c15", new StructType(ImmutableList.of(Type.INT))));
        columns.add(new Column("c16", Type.TIME));

        Schema schema = IcebergApiConverter.toIcebergApiSchema(columns);
        Assert.assertEquals("table {\n" +
                "  1: c1: required boolean\n" +
                "  2: c2: required int\n" +
                "  3: c3: required long\n" +
                "  4: c4: required float\n" +
                "  5: c5: required double\n" +
                "  6: c6: required date\n" +
                "  7: c7: required timestamp\n" +
                "  8: c8: required string\n" +
                "  9: c9: required string\n" +
                "  10: c10: required decimal(-1, -1)\n" +
                "  11: c11: required decimal(-1, -1)\n" +
                "  12: c12: required decimal(-1, -1)\n" +
                "  13: c13: required list<int>\n" +
                "  14: c14: required map<int, int>\n" +
                "  15: c15: required struct<20: col1: optional int>\n" +
                "  16: c16: required time\n" +
                "}", schema.toString());

        PartitionSpec spec = IcebergApiConverter.parsePartitionFields(schema, Lists.newArrayList("c1"));
        Assert.assertTrue(spec.isPartitioned());
        Assert.assertEquals(1, spec.fields().size());
    }