@Override
    public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath)
            throws TableNotExistException, TableNotPartitionedException, CatalogException {
        Table table = loadIcebergTable(tablePath);

        if (table.spec().isUnpartitioned()) {
            throw new TableNotPartitionedException(icebergCatalog.name(), tablePath);
        }

        Set<CatalogPartitionSpec> set = Sets.newHashSet();
        try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
            for (DataFile dataFile : CloseableIterable.transform(tasks, FileScanTask::file)) {
                Map<String, String> map = Maps.newHashMap();
                StructLike structLike = dataFile.partition();
                PartitionSpec spec = table.specs().get(dataFile.specId());
                for (int i = 0; i < structLike.size(); i++) {
                    map.put(spec.fields().get(i).name(), String.valueOf(structLike.get(i, Object.class)));
                }
                set.add(new CatalogPartitionSpec(map));
            }
        } catch (IOException e) {
            throw new CatalogException(String.format("Failed to list partitions of table %s", tablePath), e);
        }

        return Lists.newArrayList(set);
    }
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