@Override
    public SerializedMetaSpec getSerializedMetaSpec(String dbName, String tableName, long snapshotId, String serializedPredicate,
                                                    MetadataTableType metadataTableType) {
        List<RemoteMetaSplit> remoteMetaSplits = new ArrayList<>();
        IcebergTable icebergTable = (IcebergTable) getTable(dbName, tableName);
        org.apache.iceberg.Table nativeTable = icebergTable.getNativeTable();
        if (snapshotId == -1) {
            Snapshot currentSnapshot = nativeTable.currentSnapshot();
            if (currentSnapshot == null) {
                return IcebergMetaSpec.EMPTY;
            } else {
                snapshotId = nativeTable.currentSnapshot().snapshotId();
            }
        }
        Snapshot snapshot = nativeTable.snapshot(snapshotId);

        Expression predicate = Expressions.alwaysTrue();
        if (!Strings.isNullOrEmpty(serializedPredicate)) {
            predicate = SerializationUtil.deserializeFromBase64(serializedPredicate);
        }

        FileIO fileIO = nativeTable.io();
        if (fileIO instanceof IcebergCachingFileIO) {
            fileIO = ((IcebergCachingFileIO) fileIO).getWrappedIO();
        }

        String serializedTable = SerializationUtil.serializeToBase64(new SerializableTable(nativeTable, fileIO));

        if (IcebergMetaSplit.onlyNeedSingleSplit(metadataTableType)) {
            return new IcebergMetaSpec(serializedTable, List.of(IcebergMetaSplit.placeholderSplit()), false);
        }

        List<ManifestFile> dataManifests = snapshot.dataManifests(nativeTable.io());

        List<ManifestFile> matchingDataManifests = filterManifests(dataManifests, nativeTable, predicate);
        for (ManifestFile file : matchingDataManifests) {
            remoteMetaSplits.add(IcebergMetaSplit.from(file));
        }

        List<ManifestFile> deleteManifests = snapshot.deleteManifests(nativeTable.io());
        List<ManifestFile> matchingDeleteManifests = filterManifests(deleteManifests, nativeTable, predicate);
        if (metadataTableType == MetadataTableType.FILES || metadataTableType == MetadataTableType.PARTITIONS) {
            for (ManifestFile file : matchingDeleteManifests) {
                remoteMetaSplits.add(IcebergMetaSplit.from(file));
            }
            return new IcebergMetaSpec(serializedTable, remoteMetaSplits, false);
        }

        boolean loadColumnStats = enableCollectColumnStatistics(ConnectContext.get()) ||
                (!matchingDeleteManifests.isEmpty() && mayHaveEqualityDeletes(snapshot) &&
                        catalogProperties.enableDistributedPlanLoadColumnStatsWithEqDelete());

        return new IcebergMetaSpec(serializedTable, remoteMetaSplits, loadColumnStats);
    }
@Test
    public void testGetMetaSpec(@Mocked LocalMetastore localMetastore, @Mocked TemporaryTableMgr temporaryTableMgr) {
        mockedNativeTableG.newAppend().appendFile(FILE_B_5).commit();
        new MockUp<IcebergHiveCatalog>() {
            @Mock
            org.apache.iceberg.Table getTable(String dbName, String tableName) throws StarRocksConnectorException {
                return mockedNativeTableG;
            }
        };

        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(),
                new IcebergCatalogProperties(DEFAULT_CONFIG));
        ConnectContext.get().getSessionVariable().setEnableIcebergColumnStatistics(false);

        MetadataMgr metadataMgr = new MetadataMgr(localMetastore, temporaryTableMgr, null, null);
        new MockUp<MetadataMgr>() {
            @Mock
            public Optional<ConnectorMetadata> getOptionalMetadata(String catalogName) {
                return Optional.of(metadata);
            }
        };

        SerializedMetaSpec metaSpec = metadataMgr.getSerializedMetaSpec(
                "catalog", "db", "tg", -1, null, MetadataTableType.LOGICAL_ICEBERG_METADATA);
        Assert.assertTrue(metaSpec instanceof IcebergMetaSpec);
        IcebergMetaSpec icebergMetaSpec = metaSpec.cast();
        List<RemoteMetaSplit> splits = icebergMetaSpec.getSplits();
        Assert.assertFalse(icebergMetaSpec.loadColumnStats());
        Assert.assertEquals(1, splits.size());
    }