@Override
    public PartitionQuickStats buildQuickStats(ConnectorSession session, SemiTransactionalHiveMetastore metastore,
            SchemaTableName table, MetastoreContext metastoreContext, String partitionId, Iterator<HiveFileInfo> files)
    {
        requireNonNull(session);
        requireNonNull(metastore);
        requireNonNull(table);
        requireNonNull(metastoreContext);
        requireNonNull(partitionId);
        requireNonNull(files);

        if (!files.hasNext()) {
            return PartitionQuickStats.EMPTY;
        }

        // TODO: Consider refactoring storage and/or table format to the interface when we implement an ORC/Iceberg quick stats builder
        StorageFormat storageFormat;
        if (UNPARTITIONED_ID.getPartitionName().equals(partitionId)) {
            Table resolvedTable = metastore.getTable(metastoreContext, table.getSchemaName(), table.getTableName()).get();
            storageFormat = resolvedTable.getStorage().getStorageFormat();
        }
        else {
            Partition partition = metastore.getPartitionsByNames(metastoreContext, table.getSchemaName(), table.getTableName(),
                    ImmutableList.of(new PartitionNameWithVersion(partitionId, Optional.empty()))).get(partitionId).get();
            storageFormat = partition.getStorage().getStorageFormat();
        }

        if (!PARQUET_SERDE_CLASS_NAMES.contains(storageFormat.getSerDe())) {
            // Not a parquet table/partition
            return PartitionQuickStats.EMPTY;
        }

        // We want to keep the number of files we use to build quick stats bounded, so that
        // 1. We can control total file IO overhead in a measurable way
        // 2. Planning time remains bounded
        // Future work here is to sample the file list, read their stats only and extrapolate the overall stats (TODO)
        List<CompletableFuture<ParquetMetadata>> footerFetchCompletableFutures = new ArrayList<>();
        int filesCount = 0;
        while (files.hasNext()) {
            HiveFileInfo file = files.next();
            filesCount++;
            Path path = file.getPath();
            long fileSize = file.getLength();

            HiveFileContext hiveFileContext = new HiveFileContext(
                    true,
                    NO_CACHE_CONSTRAINTS,
                    Optional.empty(),
                    OptionalLong.of(fileSize),
                    OptionalLong.empty(),
                    OptionalLong.empty(),
                    file.getFileModifiedTime(),
                    false);

            HdfsContext hdfsContext = new HdfsContext(session, table.getSchemaName(), table.getTableName());
            Configuration configuration = hdfsEnvironment.getConfiguration(hdfsContext, path);

            footerFetchCompletableFutures.add(supplyAsync(() -> {
                Stopwatch footerFetchDuration = Stopwatch.createStarted();
                try (FSDataInputStream inputStream = hdfsEnvironment.getFileSystem(hdfsContext, path).openFile(path, hiveFileContext);
                        ParquetDataSource parquetDataSource = buildHdfsParquetDataSource(inputStream, path, stats)) {
                    ParquetFileMetadata parquetFileMetadata = readFooter(parquetDataSource,
                            fileSize,
                            createDecryptor(configuration, path),
                            getReadNullMaskedParquetEncryptedValue(session));

                    footerByteSizeDistribution.add(parquetFileMetadata.getMetadataSize());
                    return parquetFileMetadata.getParquetMetadata();
                }
                catch (Exception e) {
                    log.error(e);
                    throw new RuntimeException(e);
                }
                finally {
                    this.footerFetchDuration.add(footerFetchDuration.elapsed(MILLISECONDS), MILLISECONDS);
                }
            }, footerFetchExecutor));
        }

        // Record a metric about how many files were seen
        session.getRuntimeStats().addMetricValue(String.format("ParquetQuickStatsBuilder/FileCount/%s/%s", table.getTableName(), partitionId), RuntimeUnit.NONE, filesCount);
        fileCountPerPartition.add(filesCount);

        HashMap<ColumnPath, ColumnQuickStats<?>> rolledUpColStats = new HashMap<>();
        try {
            // Wait for footer reads to finish
            CompletableFuture<Void> overallCompletableFuture = CompletableFuture.allOf(footerFetchCompletableFutures.toArray(new CompletableFuture[0]));
            overallCompletableFuture.get(footerFetchTimeoutMillis, MILLISECONDS);

            for (CompletableFuture<ParquetMetadata> future : footerFetchCompletableFutures) {
                ParquetMetadata parquetMetadata = future.get();
                processColumnMetadata(parquetMetadata, rolledUpColStats);
            }
        }
        catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error(e, "Failed to read/build stats from parquet footer");
            throw new RuntimeException(e);
        }

        if (rolledUpColStats.isEmpty()) {
            return PartitionQuickStats.EMPTY;
        }
        return new PartitionQuickStats(partitionId, rolledUpColStats.values(), filesCount);
    }
@Test
    public void testStatsBuildTimeIsBoundedUsingFooterFetchTimeout()
    {
        HiveClientConfig customHiveClientConfig = new HiveClientConfig().setParquetQuickStatsFileMetadataFetchTimeout(new Duration(10, TimeUnit.MILLISECONDS));
        HdfsEnvironment mockHdfsEnvironment = new DelayingHdfsEnvironment(hdfsEnvironment, hiveClientConfig, metastoreClientConfig);

        String resourceDir = TestParquetQuickStatsBuilder.class.getClassLoader().getResource("quick_stats").toString();
        ParquetQuickStatsBuilder customParquetQuickStatsBuilder = new ParquetQuickStatsBuilder(new FileFormatDataSourceStats(), mockHdfsEnvironment, customHiveClientConfig);
        ImmutableList<HiveFileInfo> hiveFileInfos = buildHiveFileInfos(resourceDir, "tpcds_store_sales_sf_point_01", 1);

        try {
            customParquetQuickStatsBuilder.buildQuickStats(SESSION, metastore, new SchemaTableName(TEST_SCHEMA, TEST_TABLE),
                    metastoreContext, UNPARTITIONED_ID.getPartitionName(), hiveFileInfos.iterator());
        }
        catch (RuntimeException ex) {
            assertEquals(TimeoutException.class, ex.getCause().getClass());
        }
    }