@Deprecated
  public static Schema convert(TableSchema schema) {
    LogicalType schemaType = schema.toRowDataType().getLogicalType();
    Preconditions.checkArgument(
        schemaType instanceof RowType, "Schema logical type should be row type.");

    RowType root = (RowType) schemaType;
    Type converted = root.accept(new FlinkTypeToType(root));

    Schema icebergSchema = new Schema(converted.asStructType().fields());
    if (schema.getPrimaryKey().isPresent()) {
      return freshIdentifierFieldIds(icebergSchema, schema.getPrimaryKey().get().getColumns());
    } else {
      return icebergSchema;
    }
  }
@Test
  public void testConvertFlinkSchemaBaseOnIcebergSchema() {
    Schema baseSchema =
        new Schema(
            Lists.newArrayList(
                Types.NestedField.required(101, "int", Types.IntegerType.get()),
                Types.NestedField.optional(102, "string", Types.StringType.get())),
            Sets.newHashSet(101));

    TableSchema flinkSchema =
        TableSchema.builder()
            .field("int", DataTypes.INT().notNull())
            .field("string", DataTypes.STRING().nullable())
            .primaryKey("int")
            .build();
    Schema convertedSchema = FlinkSchemaUtil.convert(baseSchema, flinkSchema);
    assertThat(convertedSchema.asStruct()).isEqualTo(baseSchema.asStruct());
    assertThat(convertedSchema.identifierFieldIds()).containsExactly(101);
  }