@Override
    public List<RemoteFileInfo> getRemoteFiles(Table table, GetRemoteFilesParams params) {
        TableVersionRange version = params.getTableVersionRange();
        long snapshotId = version.end().isPresent() ? version.end().get() : -1;
        return getRemoteFiles((IcebergTable) table, snapshotId, params.getPredicate(), params.getLimit());
    }
@Test
    public void testGetRemoteFile() throws IOException {
        IcebergHiveCatalog icebergHiveCatalog = new IcebergHiveCatalog(CATALOG_NAME, new Configuration(), DEFAULT_CONFIG);
        List<Column> columns = Lists.newArrayList(new Column("k1", INT), new Column("k2", INT));
        IcebergMetadata metadata = new IcebergMetadata(CATALOG_NAME, HDFS_ENVIRONMENT, icebergHiveCatalog,
                Executors.newSingleThreadExecutor(), Executors.newSingleThreadExecutor(), null);
        IcebergTable icebergTable = new IcebergTable(1, "srTableName", CATALOG_NAME, "resource_name", "iceberg_db",
                "iceberg_table", "", columns, mockedNativeTableB, Maps.newHashMap());

        mockedNativeTableB.newAppend().appendFile(FILE_B_1).appendFile(FILE_B_2).commit();
        mockedNativeTableB.refresh();

        long snapshotId = mockedNativeTableB.currentSnapshot().snapshotId();
        ScalarOperator predicate = new BinaryPredicateOperator(BinaryType.GE,
                new ColumnRefOperator(1, INT, "k2", true), ConstantOperator.createInt(1));
        List<RemoteFileInfo> res = metadata.getRemoteFiles(icebergTable,
                GetRemoteFilesParams.newBuilder().setTableVersionRange(TableVersionRange.withEnd(Optional.of(snapshotId)))
                        .setPredicate(predicate).setFieldNames(Lists.newArrayList()).setLimit(10).build());
        IcebergRemoteFileDesc fileDesc = (IcebergRemoteFileDesc) res.get(0).getFiles().get(0);
        Assert.assertEquals(7, fileDesc.getIcebergScanTasks().stream()
                .map(x -> x.file().recordCount()).reduce(0L, Long::sum), 0.001);

        StarRocksAssert starRocksAssert = new StarRocksAssert();
        starRocksAssert.getCtx().getSessionVariable().setEnablePruneIcebergManifest(true);
        mockedNativeTableB.refresh();
        snapshotId = mockedNativeTableB.currentSnapshot().snapshotId();
        predicate = new BinaryPredicateOperator(BinaryType.EQ,
                new ColumnRefOperator(1, INT, "k2", true), ConstantOperator.createInt(2));
        res = metadata.getRemoteFiles(icebergTable,
                GetRemoteFilesParams.newBuilder().setTableVersionRange(TableVersionRange.withEnd(Optional.of(snapshotId)))
                        .setPredicate(predicate).setFieldNames(Lists.newArrayList()).setLimit(10).build());
        fileDesc = (IcebergRemoteFileDesc) res.get(0).getFiles().get(0);
        Assert.assertEquals(1, fileDesc.getIcebergScanTasks().size());
        Assert.assertEquals(3, fileDesc.getIcebergScanTasks().get(0).file().recordCount());

        PredicateSearchKey filter = PredicateSearchKey.of("db", "table", 1, null);
        Assert.assertEquals("Filter{databaseName='db', tableName='table', snapshotId=1, predicate=true}",
                filter.toString());
    }