public static TableSchema filterWatermark(TableSchema tableSchema) {
    List<WatermarkSpec> watermarkSpecs = tableSchema.getWatermarkSpecs();
    if (watermarkSpecs.isEmpty()) {
      return tableSchema;
    }

    Function<TableColumn, Boolean> filter =
        (tableColumn) -> {
          boolean isWatermark = false;
          for (WatermarkSpec spec : watermarkSpecs) {
            if (spec.getRowtimeAttribute().equals(tableColumn.getName())) {
              isWatermark = true;
              break;
            }
          }
          return !isWatermark;
        };
    return filter(tableSchema, filter).build();
  }
filter watermark due to watermark is a virtual field for now, not in mixed-format physical
table.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/FlinkSchemaUtil.java