@Override
    public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
            throws CatalogException, TableNotExistException {
        validateFlinkTable(newTable);

        Table icebergTable;
        try {
            icebergTable = loadIcebergTable(tablePath);
        } catch (TableNotExistException e) {
            if (!ignoreIfNotExists) {
                throw e;
            } else {
                return;
            }
        }

        CatalogTable table = toCatalogTable(icebergTable);

        // Currently, Flink SQL only support altering table properties.

        // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
        // comparing
        // CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
        validateTableSchemaAndPartition(table, (CatalogTable) newTable);

        Map<String, String> oldProperties = table.getOptions();
        Map<String, String> setProperties = Maps.newHashMap();

        String setLocation = null;
        String setSnapshotId = null;
        String pickSnapshotId = null;

        for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
            String key = entry.getKey();
            String value = entry.getValue();

            if (Objects.equals(value, oldProperties.get(key))) {
                continue;
            }

            if ("location".equalsIgnoreCase(key)) {
                setLocation = value;
            } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
                setSnapshotId = value;
            } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
                pickSnapshotId = value;
            } else {
                setProperties.put(key, value);
            }
        }

        oldProperties
                .keySet()
                .forEach(
                        k -> {
                            if (!newTable.getOptions().containsKey(k)) {
                                setProperties.put(k, null);
                            }
                        });

        commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
    }
Append a new level to the base namespace
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalog.java