private static void commitChanges(Table table, String setLocation, String setSnapshotId,
            String pickSnapshotId, Map<String, String> setProperties) {
        // don't allow setting the snapshot and picking a commit
        // at the same time because order is ambiguous and choosing
        // one order leads to different results
        Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
                "Cannot set the current snapshot ID and cherry-pick snapshot changes");

        if (setSnapshotId != null) {
            long newSnapshotId = Long.parseLong(setSnapshotId);
            table.manageSnapshots().setCurrentSnapshot(newSnapshotId).commit();
        }

        // if updating the table snapshot, perform that update first in case it fails
        if (pickSnapshotId != null) {
            long newSnapshotId = Long.parseLong(pickSnapshotId);
            table.manageSnapshots().cherrypick(newSnapshotId).commit();
        }

        Transaction transaction = table.newTransaction();

        if (setLocation != null) {
            transaction.updateLocation()
                    .setLocation(setLocation)
                    .commit();
        }

        if (!setProperties.isEmpty()) {
            UpdateProperties updateProperties = transaction.updateProperties();
            setProperties.forEach((k, v) -> {
                if (v == null) {
                    updateProperties.remove(k);
                } else {
                    updateProperties.set(k, v);
                }
            });
            updateProperties.commit();
        }

        transaction.commitTransaction();
    }
A Flink Catalog implementation that wraps an Iceberg {@link Catalog}.
<p>
The mapping between Flink database and Iceberg namespace:
Supplying a base namespace for a given catalog, so if you have a catalog that supports a 2-level namespace, you
would supply the first level in the catalog configuration and the second level would be exposed as Flink databases.
</p>
The Iceberg table manages its partitions by itself. The partition of the Iceberg table is independent of the
partition of Flink.

Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalog.java