@Override
  public Map<String, String> properties() {
    long changeWatermark = TablePropertyUtil.getTableWatermark(changeTable.properties());
    long baseWatermark = TablePropertyUtil.getTableWatermark(baseTable.properties());

    Map<String, String> properties = Maps.newHashMap();
    if (changeWatermark > baseWatermark) {
      baseTable
          .properties()
          .forEach(
              (k, v) -> {
                if (!TableProperties.WATERMARK_TABLE.equals(k)) {
                  properties.put(k, v);
                }
              });
      properties.put(TableProperties.WATERMARK_TABLE, String.valueOf(changeWatermark));
    } else {
      properties.putAll(baseTable.properties());
    }
    properties.put(TableProperties.WATERMARK_BASE_STORE, String.valueOf(baseWatermark));
    return ImmutableMap.copyOf(properties);
  }
Basic implementation of {@link KeyedTable}, wrapping a {@link BaseTable} and a {@link
ChangeTable}.
amoro-format-iceberg/src/main/java/org/apache/amoro/table/BasicKeyedTable.java