public static TableSchema toSchema(
      Schema icebergSchema, List<String> primaryKeys, Map<String, String> tableProperties) {
    TableSchema.Builder builder = TableSchema.builder();
    RowType rowType = org.apache.iceberg.flink.FlinkSchemaUtil.convert(icebergSchema);

    // add physical columns.
    for (RowType.RowField field : rowType.getFields()) {
      builder.field(field.getName(), TypeConversions.fromLogicalToDataType(field.getType()));
    }

    // add primary key
    if (CollectionUtils.isNotEmpty(primaryKeys)) {
      builder.primaryKey(primaryKeys.toArray(new String[0]));
    }

    Set<Integer> computeIndex = getComputeIndex(tableProperties);
    List<String> fieldNames = rowType.getFieldNames();

    // add computed columns
    for (int index : computeIndex) {
      builder.add(deserializeComputeColumn(tableProperties, index, fieldNames));
      fieldNames.add(tableProperties.get(compoundKey(FLINK_PREFIX, COMPUTED_COLUMNS, index, NAME)));
    }

    // add watermark
    if (isWatermarkValid(tableProperties)) {
      builder.watermark(deserializeWatermarkSpec(tableProperties, fieldNames));
    }
    return builder.build();
  }
Convert iceberg Schema to flink TableSchema.

@param icebergSchema
@param tableProperties
@return Flink TableSchema
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/FlinkSchemaUtil.java