@Override
    public Statistics getTableStatistics(OptimizerContext session,
                                         Table table,
                                         Map<ColumnRefOperator, Column> columns,
                                         List<PartitionKey> partitionKeys,
                                         ScalarOperator predicate,
                                         long limit,
                                         TableVersionRange version) {
        IcebergTable icebergTable = (IcebergTable) table;
        long snapshotId;
        if (version.end().isPresent()) {
            snapshotId = version.end().get();
        } else {
            Statistics.Builder statisticsBuilder = Statistics.builder();
            statisticsBuilder.setOutputRowCount(1);
            statisticsBuilder.addColumnStatistics(statisticProvider.buildUnknownColumnStatistics(columns.keySet()));
            return statisticsBuilder.build();
        }

        PredicateSearchKey key = PredicateSearchKey.of(
                icebergTable.getRemoteDbName(), icebergTable.getRemoteTableName(), snapshotId, predicate);

        triggerIcebergPlanFilesIfNeeded(key, icebergTable, predicate, limit);

        if (!session.getSessionVariable().enableIcebergColumnStatistics()) {
            List<FileScanTask> icebergScanTasks = splitTasks.get(key);
            if (icebergScanTasks == null) {
                throw new StarRocksConnectorException("Missing iceberg split task for table:[{}.{}]. predicate:[{}]",
                        icebergTable.getRemoteDbName(), icebergTable.getRemoteTableName(), predicate);
            }
            try (Timer ignored = Tracers.watchScope(EXTERNAL, "ICEBERG.calculateCardinality" + key)) {
                return statisticProvider.getCardinalityStats(columns, icebergScanTasks);
            }
        } else {
            return statisticProvider.getTableStatistics(icebergTable, columns, session, predicate, version);
        }
    }
@Test
    public void testGetTableStatistics() {
        IcebergHiveCatalog icebergHiveCatalog = new IcebergHiveCatalog(CATALOG_NAME, new Configuration(), DEFAULT_CONFIG);

        IcebergMetadata metadata = new IcebergMetadata(CATALOG_NAME, HDFS_ENVIRONMENT, icebergHiveCatalog,
                Executors.newSingleThreadExecutor(), Executors.newSingleThreadExecutor(), null);
        mockedNativeTableA.newFastAppend().appendFile(FILE_A).appendFile(FILE_A_1).commit();
        IcebergTable icebergTable = new IcebergTable(1, "srTableName", CATALOG_NAME, "resource_name", "db_name",
                "table_name", "", Lists.newArrayList(), mockedNativeTableA, Maps.newHashMap());
        Map<ColumnRefOperator, Column> colRefToColumnMetaMap = new HashMap<ColumnRefOperator, Column>();
        ColumnRefOperator columnRefOperator1 = new ColumnRefOperator(3, Type.INT, "id", true);
        ColumnRefOperator columnRefOperator2 = new ColumnRefOperator(4, Type.STRING, "data", true);
        colRefToColumnMetaMap.put(columnRefOperator1, new Column("id", Type.INT));
        colRefToColumnMetaMap.put(columnRefOperator2, new Column("data", Type.STRING));
        OptimizerContext context = new OptimizerContext(new Memo(), new ColumnRefFactory());
        Assert.assertFalse(context.getSessionVariable().enableIcebergColumnStatistics());
        Assert.assertTrue(context.getSessionVariable().enableReadIcebergPuffinNdv());
        TableVersionRange versionRange = TableVersionRange.withEnd(Optional.of(
                mockedNativeTableA.currentSnapshot().snapshotId()));
        Statistics statistics = metadata.getTableStatistics(
                context, icebergTable, colRefToColumnMetaMap, null, null, -1, versionRange);
        Assert.assertEquals(4.0, statistics.getOutputRowCount(), 0.001);
        Assert.assertEquals(2, statistics.getColumnStatistics().size());
        Assert.assertTrue(statistics.getColumnStatistic(columnRefOperator1).isUnknown());
        Assert.assertTrue(statistics.getColumnStatistic(columnRefOperator2).isUnknown());
    }