@VisibleForTesting
    static long calculateFilesSizeWithOpenCost(
            List<HiveTablePartition> partitions, JobConf jobConf, long openCost)
            throws IOException {
        long totalBytesWithWeight = 0;
        int calPartitionSizeThreadNum =
                Integer.parseInt(
                        jobConf.get(
                                HiveOptions.TABLE_EXEC_HIVE_CALCULATE_PARTITION_SIZE_THREAD_NUM
                                        .key()));
        ExecutorService executorService = null;
        try {
            executorService =
                    calPartitionSizeThreadNum == 1
                            ? newDirectExecutorService()
                            : Executors.newFixedThreadPool(calPartitionSizeThreadNum);
            List<Future<Long>> partitionFilesSizeFutures = new ArrayList<>();
            for (HiveTablePartition partition : partitions) {
                partitionFilesSizeFutures.add(
                        executorService.submit(
                                new PartitionFilesSizeCalculator(partition, openCost, jobConf)));
            }
            for (Future<Long> fileSizeFuture : partitionFilesSizeFutures) {
                try {
                    totalBytesWithWeight += fileSizeFuture.get();
                } catch (InterruptedException | ExecutionException e) {
                    throw new IOException("Fail to calculate total files' size.", e);
                }
            }
        } finally {
            if (executorService != null) {
                executorService.shutdown();
            }
        }
        return totalBytesWithWeight;
    }
@Test
    public void testCalculateFilesSize() throws Exception {
        String baseFilePath =
                Objects.requireNonNull(this.getClass().getResource("/orc/test.orc")).getPath();
        long fileSize = Paths.get(baseFilePath).toFile().length();
        File wareHouse = temporaryFolder.newFolder("testCalculateFilesSize");
        int partitionNum = 10;
        long openCost = 1;
        List<HiveTablePartition> hiveTablePartitions = new ArrayList<>();
        for (int i = 0; i < partitionNum; i++) {
            // create partition directory
            Path partitionPath = Paths.get(wareHouse.getPath(), "p_" + i);
            Files.createDirectory(partitionPath);
            // copy file to the partition directory
            Files.copy(Paths.get(baseFilePath), Paths.get(partitionPath.toString(), "t.orc"));
            StorageDescriptor sd = new StorageDescriptor();
            sd.setLocation(partitionPath.toString());
            hiveTablePartitions.add(new HiveTablePartition(sd, new Properties()));
        }
        // test calculation with one single thread
        JobConf jobConf = new JobConf();
        jobConf.set(HiveOptions.TABLE_EXEC_HIVE_CALCULATE_PARTITION_SIZE_THREAD_NUM.key(), "1");
        long totalSize =
                HiveSourceFileEnumerator.calculateFilesSizeWithOpenCost(
                        hiveTablePartitions, jobConf, openCost);
        long expectedSize = partitionNum * (fileSize + openCost);
        assertThat(totalSize).isEqualTo(expectedSize);

        // test calculation with multiple threads
        jobConf.set(HiveOptions.TABLE_EXEC_HIVE_CALCULATE_PARTITION_SIZE_THREAD_NUM.key(), "3");
        totalSize =
                HiveSourceFileEnumerator.calculateFilesSizeWithOpenCost(
                        hiveTablePartitions, jobConf, openCost);
        assertThat(totalSize).isEqualTo(expectedSize);
    }