private List<IcebergColumnInfo> getColumnList(SinkInfo sinkInfo) {
        List<StreamSinkFieldEntity> fieldList = sinkFieldMapper.selectBySinkId(sinkInfo.getId());

        // set columns
        List<IcebergColumnInfo> columnList = new ArrayList<>();
        for (StreamSinkFieldEntity field : fieldList) {
            IcebergColumnInfo column = IcebergColumnInfo.getFromJson(field.getExtParams());
            column.setFieldName(field.getFieldName());
            column.setFieldType(field.getFieldType());
            column.setFieldComment(field.getFieldComment());
            column.setRequired(field.getIsRequired() != null && field.getIsRequired() > 0);
            columnList.add(column);
        }

        return columnList;
    }
Create iceberg table according to the sink config
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/iceberg/IcebergResourceOperator.java