private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
        ImmutableList.Builder<String> partitionKeysBuilder = ImmutableList.builder();
        for (PartitionField field : spec.fields()) {
            if (field.transform().isIdentity()) {
                partitionKeysBuilder.add(icebergSchema.findColumnName(field.sourceId()));
            } else {
                // Not created by Flink SQL.
                // For compatibility with iceberg tables, return empty.
                // TODO modify this after Flink support partition transform.
                return Collections.emptyList();
            }
        }
        return partitionKeysBuilder.build();
    }
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