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

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

        try {
            CompletableFuture<Map<ConnectorTableColumnKey, Optional<Histogram>>> result =
                    connectorHistogramCache.getAll(cacheKeys);
            if (result.isDone()) {
                Map<ConnectorTableColumnKey, Optional<Histogram>> realResult = result.get();

                Map<String, Histogram> histogramStats = Maps.newHashMap();
                for (String columnName : columns) {
                    Optional<Histogram> histogramStatistics =
                            realResult.getOrDefault(new ConnectorTableColumnKey(table.getUUID(), columnName), Optional.empty());
                    histogramStatistics.ifPresent(histogram -> histogramStats.put(columnName, histogram));
                }
                return histogramStats;
            } else {
                return Maps.newHashMap();
            }
        } catch (Exception e) {
            LOG.warn("Failed to execute getConnectorHistogramStatistics", e);
            return Maps.newHashMap();
        }
    }
@Test
    public void testGetConnectorHistogramStatistics(@Mocked AsyncLoadingCache<ConnectorTableColumnKey, Optional<Histogram>>
                                                    histogramCache) {
        Table table = connectContext.getGlobalStateMgr().getMetadataMgr().getTable("hive0", "partitioned_db", "t1");
        ConnectorTableColumnKey key = new ConnectorTableColumnKey("hive0.partitioned_db.t1.1234", "c1");
        new Expectations() {
            {
                histogramCache.getAll((Iterable<? extends ConnectorTableColumnKey>) any);
                result = CompletableFuture.completedFuture(ImmutableMap.of(key, Optional.empty()));
                minTimes = 0;
            }
        };
        CachedStatisticStorage cachedStatisticStorage = new CachedStatisticStorage();
        Map<String, Histogram> histogramMap =
                cachedStatisticStorage.getConnectorHistogramStatistics(table, ImmutableList.of("c1"));
        Assert.assertEquals(0, histogramMap.size());
    }