void createIcebergTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
            throws CatalogException, TableAlreadyExistException {
        validateFlinkTable(table);

        Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
        PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);

        ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
        String location = null;
        for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
            if ("location".equalsIgnoreCase(entry.getKey())) {
                location = entry.getValue();
            } else {
                properties.put(entry.getKey(), entry.getValue());
            }
        }

        try {
            icebergCatalog.createTable(
                    toIdentifier(tablePath),
                    icebergSchema,
                    spec,
                    location,
                    properties.build());
        } catch (AlreadyExistsException e) {
            if (!ignoreIfExists) {
                throw new TableAlreadyExistException(getName(), tablePath, e);
            }
        }
    }
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