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();
    }
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