static CatalogTable toCatalogTable(Table table) {
        TableSchema schema = FlinkSchemaUtil.toSchema(table.schema());
        List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());

        // NOTE: We can not create a IcebergCatalogTable extends CatalogTable, because Flink optimizer
        // may use
        // CatalogTableImpl to copy a new catalog table.
        // Let's re-loading table from Iceberg catalog when creating source/sink operators.
        // Iceberg does not have Table comment, so pass a null (Default comment value in Flink).
        return new CatalogTableImpl(schema, partitionKeys, table.properties(), null);
    }
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