private static void validateTableSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) {
        TableSchema ts1 = ct1.getSchema();
        TableSchema ts2 = ct2.getSchema();
        boolean equalsPrimary = false;

        if (ts1.getPrimaryKey().isPresent() && ts2.getPrimaryKey().isPresent()) {
            equalsPrimary =
                    Objects.equals(ts1.getPrimaryKey().get().getType(), ts2.getPrimaryKey().get().getType())
                            && Objects.equals(
                                    ts1.getPrimaryKey().get().getColumns(), ts2.getPrimaryKey().get().getColumns());
        } else if (!ts1.getPrimaryKey().isPresent() && !ts2.getPrimaryKey().isPresent()) {
            equalsPrimary = true;
        }

        if (!(Objects.equals(ts1.getTableColumns(), ts2.getTableColumns())
                && Objects.equals(ts1.getWatermarkSpecs(), ts2.getWatermarkSpecs())
                && equalsPrimary)) {
            throw new UnsupportedOperationException("Altering schema is not supported yet.");
        }

        if (!ct1.getPartitionKeys().equals(ct2.getPartitionKeys())) {
            throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
        }
    }
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