public Statistics getTableStatistics(IcebergTable icebergTable,
                                         Map<ColumnRefOperator, Column> colRefToColumnMetaMap,
                                         OptimizerContext session,
                                         ScalarOperator predicate,
                                         TableVersionRange version) {
        Table nativeTable = icebergTable.getNativeTable();
        Statistics.Builder statisticsBuilder = Statistics.builder();
        String uuid = icebergTable.getUUID();
        if (version.end().isPresent()) {
            Set<Integer> primitiveColumnsFieldIds = nativeTable.schema().columns().stream()
                    .filter(column -> column.type().isPrimitiveType())
                    .map(Types.NestedField::fieldId).collect(Collectors.toSet());
            Map<Integer, Long> colIdToNdvs = new HashMap<>();
            if (session != null && session.getSessionVariable().enableReadIcebergPuffinNdv()) {
                colIdToNdvs = readNumDistinctValues(icebergTable, primitiveColumnsFieldIds, version);
                if (uuidToPartitionFieldIdToValues.containsKey(uuid) && !uuidToPartitionFieldIdToValues.get(uuid).isEmpty()) {
                    HashMultimap<Integer, Object> partitionFieldIdToValue = uuidToPartitionFieldIdToValues.get(uuid);
                    Map<Integer, Long> partitionSourceIdToNdv = new HashMap<>();
                    for (PartitionField partitionField : nativeTable.spec().fields()) {
                        int sourceId = partitionField.sourceId();
                        int fieldId = partitionField.fieldId();
                        if (partitionFieldIdToValue.containsKey(fieldId)) {
                            partitionSourceIdToNdv.put(sourceId, (long) partitionFieldIdToValue.get(fieldId).size());
                        }
                    }
                    colIdToNdvs.putAll(partitionSourceIdToNdv);
                }
            }

            PredicateSearchKey key = PredicateSearchKey.of(icebergTable.getRemoteDbName(), icebergTable.getRemoteTableName(),
                    version.end().get(), predicate);
            IcebergFileStats icebergFileStats;
            if (!icebergFileStatistics.containsKey(key)) {
                icebergFileStats = new IcebergFileStats(1);
            } else {
                icebergFileStats = icebergFileStatistics.get(key);
            }

            statisticsBuilder.setOutputRowCount(icebergFileStats.getRecordCount());
            statisticsBuilder.addColumnStatistics(buildColumnStatistics(
                    nativeTable, colRefToColumnMetaMap, icebergFileStats, colIdToNdvs));
        } else {
            // empty table
            statisticsBuilder.setOutputRowCount(1);
            statisticsBuilder.addColumnStatistics(buildUnknownColumnStatistics(colRefToColumnMetaMap.keySet()));
        }
        return statisticsBuilder.build();
    }
@Test
    public void testGetEmptyTableStatistics() {
        IcebergStatisticProvider statisticProvider = new IcebergStatisticProvider();
        IcebergTable icebergTable = new IcebergTable(1, "srTableName", "iceberg_catalog", "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));
        Statistics statistics = statisticProvider.getTableStatistics(icebergTable, colRefToColumnMetaMap,
                null, null, TableVersionRange.empty());
        Assert.assertEquals(1.0, statistics.getOutputRowCount(), 0.001);
    }