@Override
    public List<PartitionInfo> getPartitions(Table table, List<String> partitionNames) {
        Map<String, Partition> partitionMap = Maps.newHashMap();
        IcebergTable icebergTable = (IcebergTable) table;
        PartitionsTable partitionsTable = (PartitionsTable) MetadataTableUtils.
                createMetadataTableInstance(icebergTable.getNativeTable(), org.apache.iceberg.MetadataTableType.PARTITIONS);

        if (icebergTable.isUnPartitioned()) {
            try (CloseableIterable<FileScanTask> tasks = partitionsTable.newScan().planFiles()) {
                for (FileScanTask task : tasks) {
                    // partitionsTable Table schema :
                    // record_count,
                    // file_count,
                    // total_data_file_size_in_bytes,
                    // position_delete_record_count,
                    // position_delete_file_count,
                    // equality_delete_record_count,
                    // equality_delete_file_count,
                    // last_updated_at,
                    // last_updated_snapshot_id
                    CloseableIterable<StructLike> rows = task.asDataTask().rows();
                    for (StructLike row : rows) {
                        // Get the last updated time of the table according to the table schema
                        long lastUpdated = -1;
                        try {
                            lastUpdated = row.get(7, Long.class);
                        } catch (NullPointerException e) {
                            LOG.error("The table [{}] snapshot [{}] has been expired",
                                    icebergTable.getRemoteDbName(), icebergTable.getRemoteTableName(), e);
                        }
                        Partition partition = new Partition(lastUpdated);
                        return ImmutableList.of(partition);
                    }
                }
                // for empty table, use -1 as last updated time
                return ImmutableList.of(new Partition(-1));
            } catch (IOException e) {
                throw new StarRocksConnectorException("Failed to get partitions for table: " + table.getName(), e);
            }
        } else {
            // For partition table, we need to get all partitions from PartitionsTable.
            try (CloseableIterable<FileScanTask> tasks = partitionsTable.newScan().planFiles()) {
                for (FileScanTask task : tasks) {
                    // partitionsTable Table schema :
                    // partition,
                    // spec_id,
                    // record_count,
                    // file_count,
                    // total_data_file_size_in_bytes,
                    // position_delete_record_count,
                    // position_delete_file_count,
                    // equality_delete_record_count,
                    // equality_delete_file_count,
                    // last_updated_at,
                    // last_updated_snapshot_id
                    CloseableIterable<StructLike> rows = task.asDataTask().rows();
                    for (StructLike row : rows) {
                        // Get the partition data/spec id/last updated time according to the table schema
                        StructProjection partitionData = row.get(0, StructProjection.class);
                        int specId = row.get(1, Integer.class);
                        PartitionSpec spec = icebergTable.getNativeTable().specs().get(specId);
                        String partitionName =
                                PartitionUtil.convertIcebergPartitionToPartitionName(spec, partitionData);

                        long lastUpdated = -1;
                        try {
                            lastUpdated = row.get(9, Long.class);
                        } catch (NullPointerException e) {
                            LOG.error("The table [{}.{}] snapshot [{}] has been expired",
                                    icebergTable.getRemoteDbName(), icebergTable.getRemoteTableName(), partitionName, e);
                        }
                        Partition partition = new Partition(lastUpdated);
                        partitionMap.put(partitionName, partition);
                    }
                }
            } catch (IOException e) {
                throw new StarRocksConnectorException("Failed to get partitions for table: " + table.getName(), e);
            }
        }
        ImmutableList.Builder<PartitionInfo> partitions = ImmutableList.builder();
        partitionNames.forEach(partitionName -> partitions.add(partitionMap.get(partitionName)));
        return partitions.build();
    }
@Test
    public void testGetPartitions1() {
        mockedNativeTableB.newAppend().appendFile(FILE_B_1).appendFile(FILE_B_2).commit();

        IcebergHiveCatalog icebergHiveCatalog = new IcebergHiveCatalog(CATALOG_NAME, new Configuration(), DEFAULT_CONFIG);
        CachingIcebergCatalog cachingIcebergCatalog = new CachingIcebergCatalog(CATALOG_NAME, icebergHiveCatalog,
                DEFAULT_CATALOG_PROPERTIES, Executors.newSingleThreadExecutor());
        IcebergMetadata metadata = new IcebergMetadata(CATALOG_NAME, HDFS_ENVIRONMENT, cachingIcebergCatalog,
                Executors.newSingleThreadExecutor(), Executors.newSingleThreadExecutor(), null);

        IcebergTable icebergTable = new IcebergTable(1, "srTableName", CATALOG_NAME,
                "resource_name", "db",
                "table", "", Lists.newArrayList(), mockedNativeTableB, Maps.newHashMap());

        List<PartitionInfo> partitions = metadata.getPartitions(icebergTable, ImmutableList.of("k2=2", "k2=3"));
        Assert.assertEquals(2, partitions.size());
    }