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(enabled = false, invocationCount = 3)
    public void testConcurrentFetchForSamePartition()
            throws ExecutionException, InterruptedException
    {
        QuickStatsBuilder longRunningQuickStatsBuilderMock = (session, metastore, table, metastoreContext, partitionId, files) -> {
            // Sleep for 50ms to simulate a long-running quick stats call
            sleepUninterruptibly(50, MILLISECONDS);
            return mockPartitionQuickStats;
        };

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

        List<String> testPartitions = ImmutableList.of("partition1", "partition2", "partition3");

        {
            // Build a session where an inline quick stats build will occur for the 1st query that initiates the build,
            // but subsequent queries will NOT wait

            ConnectorSession session = getSession("600ms", "0ms");
            // Execute two concurrent calls for the same partitions; wait for them to complete
            CompletableFuture<Map<String, PartitionStatistics>> future1 = supplyAsync(() -> quickStatsProvider.getQuickStats(session, metastoreMock,
                    new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitions), commonPool());

            CompletableFuture<Map<String, PartitionStatistics>> future2 = supplyAsync(() -> quickStatsProvider.getQuickStats(session, metastoreMock,
                    new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitions), commonPool());

            allOf(future1, future2).join();

            Map<String, PartitionStatistics> quickStats1 = future1.get();
            Map<String, PartitionStatistics> quickStats2 = future2.get();

            // Both PartitionStatistics have stats for all partitions
            assertEquals(quickStats1.entrySet().size(), testPartitions.size());
            assertTrue(quickStats1.keySet().containsAll(testPartitions));
            assertEquals(quickStats2.entrySet().size(), testPartitions.size());
            assertTrue(quickStats2.keySet().containsAll(testPartitions));

            // For the same partition, we will observe that one of them has the expected partition stats and the other one is empty
            for (String testPartition : testPartitions) {
                PartitionStatistics partitionStatistics1 = quickStats1.get(testPartition);
                PartitionStatistics partitionStatistics2 = quickStats2.get(testPartition);

                if (partitionStatistics1.equals(empty())) {
                    assertEquals(partitionStatistics2, expectedPartitionStats);
                }
                else if (partitionStatistics2.equals(empty())) {
                    assertEquals(partitionStatistics1, expectedPartitionStats);
                }
                else {
                    fail(String.format("For [%s] one of the partitions stats was expected to be empty. Actual partitionStatistics1 [%s], partitionStatistics2 [%s]",
                            testPartition, partitionStatistics1, partitionStatistics2));
                }
            }

            // Future calls for the same partitions will return from cached partition stats with valid values
            Map<String, PartitionStatistics> quickStats = quickStatsProvider.getQuickStats(session, metastoreMock,
                    new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitions);

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

        {
            // Build a session where an inline quick stats build will occur for the 1st query that initiates the build,
            // and subsequent queries will wait for this inline build to finish too

            ConnectorSession session = getSession("300ms", "300ms");
            // Execute two concurrent calls for the same partitions; wait for them to complete
            CompletableFuture<Map<String, PartitionStatistics>> future1 = supplyAsync(() -> quickStatsProvider.getQuickStats(session, metastoreMock,
                    new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitions), commonPool());

            CompletableFuture<Map<String, PartitionStatistics>> future2 = supplyAsync(() -> quickStatsProvider.getQuickStats(session, metastoreMock,
                    new SchemaTableName(TEST_SCHEMA, TEST_TABLE), metastoreContext, testPartitions), commonPool());

            allOf(future1, future2).join();

            Map<String, PartitionStatistics> quickStats1 = future1.get();
            Map<String, PartitionStatistics> quickStats2 = future2.get();

            assertEquals(quickStats1, quickStats2); // Both PartitionStatistics have exactly same stats
            assertTrue(quickStats1.keySet().containsAll(testPartitions)); // Stats for all partitions are present
            // None of the partition stats are empty, they are all the mocked value
            assertTrue(quickStats1.values().stream().allMatch(x -> x.equals(convertToPartitionStatistics(mockPartitionQuickStats))));
        }
    }