public Map<String, PartitionStatistics> getQuickStats(ConnectorSession session, SemiTransactionalHiveMetastore metastore, SchemaTableName table,
            MetastoreContext metastoreContext, List<String> partitionIds)
    {
        if (!isQuickStatsEnabled(session)) {
            return partitionIds.stream().collect(toMap(k -> k, v -> empty()));
        }

        CompletableFuture<PartitionStatistics>[] partitionQuickStatCompletableFutures = new CompletableFuture[partitionIds.size()];
        for (int counter = 0; counter < partitionIds.size(); counter++) {
            String partitionId = partitionIds.get(counter);
            partitionQuickStatCompletableFutures[counter] = supplyAsync(() -> getQuickStats(session, metastore, table, metastoreContext, partitionId), backgroundFetchExecutor);
        }

        try {
            // Wait for all the partitions to get their quick stats
            // If this query is reading a partition for which we do not already have cached quick stats,
            // we will block the execution of the query until the stats are fetched for all such partitions,
            // or we time out waiting for the fetch
            allOf(partitionQuickStatCompletableFutures).get(getQuickStatsInlineBuildTimeoutMillis(session), MILLISECONDS);
        }
        catch (InterruptedException | ExecutionException e) {
            log.error(e);
            throw new RuntimeException(e);
        }
        catch (TimeoutException e) {
            log.warn(e, "Timeout while building quick stats");
        }

        ImmutableMap.Builder<String, PartitionStatistics> result = ImmutableMap.builder();
        for (int counter = 0; counter < partitionQuickStatCompletableFutures.length; counter++) {
            String partitionId = partitionIds.get(counter);
            CompletableFuture<PartitionStatistics> future = partitionQuickStatCompletableFutures[counter];
            if (future.isDone() && !future.isCancelled() && !future.isCompletedExceptionally()) {
                try {
                    result.put(partitionId, future.get());
                }
                catch (InterruptedException | ExecutionException e) {
                    // This should not happen because we checked that the future was completed successfully
                    log.error(e, "Failed to get value for a quick stats future which was completed successfully");
                    throw new RuntimeException(e);
                }
            }
            else {
                // If a future did not finish, or finished exceptionally, we do not add it to the results
                // A new query for the same partition could trigger a successful quick stats fetch for this partition
                result.put(partitionId, empty());
            }
        }

        return result.build();
    }
@Test
    public void testReadThruCaching()
    {
        QuickStatsBuilder quickStatsBuilderMock = (session, metastore, table, metastoreContext, partitionId, files) -> mockPartitionQuickStats;

        QuickStatsProvider quickStatsProvider = new QuickStatsProvider(hdfsEnvironment, directoryListerMock, hiveClientConfig, new NamenodeStats(),
                ImmutableList.of(quickStatsBuilderMock));

        // Execute
        ImmutableList<String> testPartitions1 = ImmutableList.of("partition1", "partition2", "partition3");
        Map<String, PartitionStatistics> quickStats = quickStatsProvider.getQuickStats(SESSION, metastoreMock,
                new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitions1);

        // Verify only one call was made for each test partition
        assertEquals(quickStats.entrySet().size(), testPartitions1.size());
        assertTrue(quickStats.keySet().containsAll(testPartitions1));
        quickStats.values().forEach(ps -> assertEquals(ps, expectedPartitionStats));

        // For subsequent calls for the same partitions that are already cached, no new calls are mode to the quick stats builder
        quickStatsProvider.getQuickStats(SESSION, metastoreMock,
                new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitions1);

        // For subsequent calls with a mix of old and new partitions, we only see calls to the quick stats builder for the new partitions
        ImmutableList<String> testPartitions2 = ImmutableList.of("partition4", "partition5", "partition6");
        ImmutableList<String> testPartitionsMix = ImmutableList.<String>builder().addAll(testPartitions1).addAll(testPartitions2).build();

        quickStats = quickStatsProvider.getQuickStats(SESSION, metastoreMock,
                new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitionsMix);
        assertEquals(quickStats.entrySet().size(), testPartitionsMix.size());
        assertTrue(quickStats.keySet().containsAll(testPartitionsMix));
        quickStats.values().forEach(ps -> assertEquals(ps, expectedPartitionStats));
    }