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