presto-accumulo/src/main/java/io/prestosql/plugin/accumulo/index/ColumnCardinalityCache.java/CardinalityCacheLoader/loadAll
class CardinalityCacheLoader:        @Override
        public Map<CacheKey, Long> loadAll(Iterable<? extends CacheKey> keys)
                throws Exception
        {
            int size = Iterables.size(keys);
            if (size == 0) {
                return ImmutableMap.of();
            }

            LOG.debug("Loading %s exact ranges from Accumulo", size);

            // In order to simplify the implementation, we are making a (safe) assumption
            // that the CacheKeys will all contain the same combination of schema/table/family/qualifier
            // This is asserted with the below implementation error just to make sure
            CacheKey anyKey = stream(keys).findAny().get();
            if (stream(keys).anyMatch(k -> !k.getSchema().equals(anyKey.getSchema()) || !k.getTable().equals(anyKey.getTable()) || !k.getFamily().equals(anyKey.getFamily()) || !k.getQualifier().equals(anyKey.getQualifier()))) {
                throw new PrestoException(FUNCTION_IMPLEMENTATION_ERROR, "loadAll called with a non-homogeneous collection of cache keys");
            }

            Map<Range, CacheKey> rangeToKey = stream(keys).collect(Collectors.toMap(CacheKey::getRange, Function.identity()));
            LOG.debug("rangeToKey size is %s", rangeToKey.size());

            // Get metrics table name and the column family for the scanner
            String metricsTable = getMetricsTableName(anyKey.getSchema(), anyKey.getTable());
            Text columnFamily = new Text(getIndexColumnFamily(anyKey.getFamily().getBytes(UTF_8), anyKey.getQualifier().getBytes(UTF_8)).array());

            BatchScanner scanner = connector.createBatchScanner(metricsTable, anyKey.getAuths(), 10);
            try {
                scanner.setRanges(stream(keys).map(CacheKey::getRange).collect(Collectors.toList()));
                scanner.fetchColumn(columnFamily, CARDINALITY_CQ_AS_TEXT);

                // Create a new map to hold our cardinalities for each range, returning a default of
                // Zero for each non-existent Key
                Map<CacheKey, Long> rangeValues = new HashMap<>();
                stream(keys).forEach(key -> rangeValues.put(key, 0L));

                for (Entry<Key, Value> entry : scanner) {
                    rangeValues.put(rangeToKey.get(Range.exact(entry.getKey().getRow())), parseLong(entry.getValue().toString()));
                }

                return rangeValues;
            }
            finally {
                if (scanner != null) {
                    scanner.close();
                }
            }
        }