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

        // get Statistics Table column info, just return default column statistics
        if (StatisticUtils.statisticTableBlackListCheck(table.getId())) {
            return getDefaultConnectorTableStatistics(columns);
        }

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

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

        try {
            CompletableFuture<Map<ConnectorTableColumnKey, Optional<ConnectorTableColumnStats>>> result =
                    connectorTableCachedStatistics.getAll(cacheKeys);
            if (result.isDone()) {
                List<ConnectorTableColumnStats> columnStatistics = new ArrayList<>();
                Map<ConnectorTableColumnKey, Optional<ConnectorTableColumnStats>> realResult;
                realResult = result.get();
                for (String column : columns) {
                    Optional<ConnectorTableColumnStats> columnStatistic =
                            realResult.getOrDefault(new ConnectorTableColumnKey(table.getUUID(), column), Optional.empty());
                    if (columnStatistic.isPresent()) {
                        columnStatistics.add(columnStatistic.get());
                    } else {
                        columnStatistics.add(ConnectorTableColumnStats.unknown());
                    }
                }
                return columnStatistics;
            } else {
                return getDefaultConnectorTableStatistics(columns);
            }
        } catch (Exception e) {
            LOG.warn("Failed to execute connectorTableCachedStatistics.getAll", e);
            return getDefaultConnectorTableStatistics(columns);
        }
    }
@Test
    public void testGetConnectorTableStatistics() throws ExecutionException, InterruptedException {
        AsyncLoadingCache<ConnectorTableColumnKey, Optional<ConnectorTableColumnStats>>
                connectorTableCachedStatistics =
                Caffeine.newBuilder().expireAfterWrite(Config.statistic_update_interval_sec * 2, TimeUnit.SECONDS)
                        .refreshAfterWrite(Config.statistic_update_interval_sec, TimeUnit.SECONDS)
                        .maximumSize(Config.statistic_cache_columns)
                        .buildAsync(new ConnectorColumnStatsCacheLoader());
        new MockUp<ConnectorColumnStatsCacheLoader>() {
            @Mock
            public List<TStatisticData> queryStatisticsData(ConnectContext context, String tableUUID,
                                                            List<String> columns) {
                TStatisticData data1 = new TStatisticData();
                data1.setColumnName("c1");
                data1.setRowCount(5);
                data1.setDataSize(100);
                data1.setCountDistinct(5);
                data1.setNullCount(0);
                data1.setMin("0");
                data1.setMax("10");

                TStatisticData data2 = new TStatisticData();
                data2.setColumnName("c2");
                data2.setRowCount(5);
                data2.setDataSize(100);
                data2.setCountDistinct(5);
                data2.setNullCount(0);
                data2.setMin("a");
                data2.setMax("z");

                TStatisticData data3 = new TStatisticData();
                data3.setColumnName("c3");
                data3.setRowCount(5);
                data3.setDataSize(100);
                data3.setCountDistinct(5);
                data3.setNullCount(0);
                data3.setMin("x");
                data3.setMax("y");

                return ImmutableList.of(data1, data2, data3);
            }
        };

        new MockUp<StatisticsUtils>() {
            @Mock
            public Table getTableByUUID(String tableUUID) {
                return connectContext.getGlobalStateMgr().getMetadataMgr().
                        getTable("hive0", "partitioned_db", "t1");
            }

        };

        List<ConnectorTableColumnKey> cacheKeys = ImmutableList.of(
                new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234", "c1"),
                new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234", "c2"),
                new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234", "c3"));
        CompletableFuture<Map<ConnectorTableColumnKey, Optional<ConnectorTableColumnStats>>> future =
                connectorTableCachedStatistics.getAll(cacheKeys);
        Map<ConnectorTableColumnKey, Optional<ConnectorTableColumnStats>> result = future.get();

        Assert.assertEquals(3, result.size());
        Assert.assertEquals(5, result.get(new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234",
                        "c1")).get().getRowCount());
        Assert.assertEquals(20, result.get(new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234",
                        "c1")).get().getColumnStatistic().getAverageRowSize(), 0.0001);
        Assert.assertEquals(10, result.get(new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234",
                        "c1")).get().getColumnStatistic().getMaxValue(), 0.0001);
        Assert.assertEquals(0, result.get(new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234",
                        "c1")).get().getColumnStatistic().getMinValue(), 0.0001);
    }