@Override
    public List<ConnectorTableColumnStats> getConnectorTableStatisticsSync(Table table, List<String> columns) {
        Preconditions.checkState(table != null);

        if (!StatisticUtils.checkStatisticTableStateNormal()) {
            return getDefaultConnectorTableStatistics(columns);
        }

        List<ConnectorTableColumnKey> cacheKeys = new ArrayList<>();
        for (String column : columns) {
            cacheKeys.add(new ConnectorTableColumnKey(table.getUUID(), column));
        }

        try {
            Map<ConnectorTableColumnKey, Optional<ConnectorTableColumnStats>> result =
                    connectorTableCachedStatistics.synchronous().getAll(cacheKeys);
            List<ConnectorTableColumnStats> columnStatistics = new ArrayList<>();

            for (String column : columns) {
                Optional<ConnectorTableColumnStats> columnStatistic =
                        result.getOrDefault(new ConnectorTableColumnKey(table.getUUID(), column), Optional.empty());
                if (columnStatistic.isPresent()) {
                    columnStatistics.add(columnStatistic.get());
                } else {
                    columnStatistics.add(ConnectorTableColumnStats.unknown());
                }
            }
            return columnStatistics;
        } catch (Exception e) {
            LOG.warn("Failed to execute getConnectorTableStatisticsSync", e);
            return getDefaultConnectorTableStatistics(columns);
        }
    }
@Test
    public void testGetConnectorTableStatisticsSync(
            @Mocked AsyncLoadingCache<ConnectorTableColumnKey,
                    Optional<ConnectorTableColumnStats>> connectorTableCachedStatistics,
            @Mocked LoadingCache<ConnectorTableColumnKey,
                    Optional<ConnectorTableColumnStats>> connectorTableTableSyncCachedStatistics) {
        Table table = connectContext.getGlobalStateMgr().getMetadataMgr().getTable("hive0", "partitioned_db", "t1");
        List<ConnectorTableColumnKey> cacheKeys =
                ImmutableList.of(new ConnectorTableColumnKey(table.getUUID(), "c1"),
                        new ConnectorTableColumnKey(table.getUUID(), "c2"));

        Map<ConnectorTableColumnKey, Optional<ConnectorTableColumnStats>> columnKeyOptionalMap = Maps.newHashMap();
        columnKeyOptionalMap.put(new ConnectorTableColumnKey(table.getUUID(), "c1"),
                Optional.of(new ConnectorTableColumnStats(
                        new ColumnStatistic(0, 10, 0, 20, 5), 5)));
        columnKeyOptionalMap.put(new ConnectorTableColumnKey(table.getUUID(), "c2"),
                Optional.of(new ConnectorTableColumnStats(
                        new ColumnStatistic(0, 100, 0, 200, 50), 50)));

        new MockUp<StatisticUtils>() {
            @Mock
            public boolean checkStatisticTableStateNormal() {
                return true;
            }
        };

        CachedStatisticStorage cachedStatisticStorage = new CachedStatisticStorage();
        List<ConnectorTableColumnStats> connectorColumnStatistics = cachedStatisticStorage.
                getConnectorTableStatisticsSync(table, ImmutableList.of("c1", "c2"));
        Assert.assertEquals(2, connectorColumnStatistics.size());

        new MockUp<StatisticUtils>() {
            @Mock
            public boolean checkStatisticTableStateNormal() {
                return false;
            }
        };
        connectorColumnStatistics = cachedStatisticStorage.
                getConnectorTableStatisticsSync(table, ImmutableList.of("c1", "c2"));
        Assert.assertEquals(2, connectorColumnStatistics.size());
        Assert.assertTrue(connectorColumnStatistics.get(0).getColumnStatistic().isUnknown());
        Assert.assertTrue(connectorColumnStatistics.get(1).getColumnStatistic().isUnknown());
    }