public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM =
        key("table.exec.hive.infer-source-parallelism")
                .defaultValue(true)
                .withDescription(
                        "If is false, parallelism of source are set by config.\n" +
                        "If is true, source parallelism is inferred according to splits number.\n");

public static final ConfigOption<Integer> TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM_MAX =
        key("table.exec.hive.infer-source-parallelism.max")
                .defaultValue(1000)
                .withDescription("Sets max infer parallelism for source operator.");

/**
 * A utility class to calculate parallelism for Hive connector considering various factors.
 */
class HiveParallelismInference {

	private static final Logger LOG = LoggerFactory.getLogger(HiveParallelismInference.class);

	private final ObjectPath tablePath;
	private final boolean infer;
	private final int inferMaxParallelism;

	private int parallelism;

	HiveParallelismInference(ObjectPath tablePath, ReadableConfig flinkConf) {
		this.tablePath = tablePath;
        // 获取 table.exec.hive.infer-source-parallelism 配置并赋值，
		this.infer = flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM);
		// 获取 table.exec.hive.infer-source-parallelism.max 配置并赋值
        this.inferMaxParallelism = flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM_MAX);
		Preconditions.checkArgument(
			inferMaxParallelism >= 1,
			HiveOptions.TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
        // 获取 table.exec.resource.default-parallelism 配置
		this.parallelism = flinkConf.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
	}

	/**
	 * Apply limit to calculate the parallelism.
	 * Here limit is the limit in query <code>SELECT * FROM xxx LIMIT [limit]</code>.
	 */
	int limit(Long limit) {
		if (limit != null) {
			parallelism = Math.min(parallelism, (int) (limit / 1000));
		}

		// make sure that parallelism is at least 1
		return Math.max(1, parallelism);
	}

    //根据
	/**
	 * Infer parallelism by number of files and number of splits.
	 * If {@link HiveOptions#TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM} is not set this method does nothing.
	 */
	HiveParallelismInference infer(
			SupplierWithException<Integer, IOException> numFiles,
			SupplierWithException<Integer, IOException> numSplits) {
        //如果设置 table.exec.hive.infer-source-parallelism 为 false，则直接跳过了
		if (!infer) {
			return this;
		}

		try {
			// `createInputSplits` is costly,
			// so we try to avoid calling it by first checking the number of files
			// which is the lower bound of the number of splits
			int lowerBound = logRunningTime("getNumFiles", numFiles);
			if (lowerBound >= inferMaxParallelism) {
				parallelism = inferMaxParallelism;
				return this;
			}

			int splitNum = logRunningTime("createInputSplits", numSplits);
			parallelism = Math.min(splitNum, inferMaxParallelism);
		} catch (IOException e) {
			throw new FlinkHiveException(e);
		}
		return this;
	}

	private int logRunningTime(
			String operationName, SupplierWithException<Integer, IOException> supplier) throws IOException {
		long startTimeMillis = System.currentTimeMillis();
		int result = supplier.get();
		LOG.info(
			"Hive source({}}) {} use time: {} ms, result: {}",
			tablePath,
			operationName,
			System.currentTimeMillis() - startTimeMillis,
			result);
		return result;
	}
}

public static int getNumFiles(List<HiveTablePartition> partitions, JobConf jobConf) throws IOException {
    int numFiles = 0;
    FileSystem fs = null;
    for (HiveTablePartition partition : partitions) {
        StorageDescriptor sd = partition.getStorageDescriptor();
        org.apache.hadoop.fs.Path inputPath = new org.apache.hadoop.fs.Path(sd.getLocation());
        if (fs == null) {
            fs = inputPath.getFileSystem(jobConf);
        }
        // it's possible a partition exists in metastore but the data has been removed
        if (!fs.exists(inputPath)) {
            continue;
        }
        numFiles += fs.listStatus(inputPath).length;
    }
    return numFiles;
}

public static List<HiveSourceSplit> createInputSplits(
        int minNumSplits,
        List<HiveTablePartition> partitions,
        JobConf jobConf) throws IOException {
    List<HiveSourceSplit> hiveSplits = new ArrayList<>();
    FileSystem fs = null;
    for (HiveTablePartition partition : partitions) {
        StorageDescriptor sd = partition.getStorageDescriptor();
        org.apache.hadoop.fs.Path inputPath = new org.apache.hadoop.fs.Path(sd.getLocation());
        if (fs == null) {
            fs = inputPath.getFileSystem(jobConf);
        }
        // it's possible a partition exists in metastore but the data has been removed
        if (!fs.exists(inputPath)) {
            continue;
        }
        InputFormat format;
        try {
            format = (InputFormat)
                    Class.forName(sd.getInputFormat(), true, Thread.currentThread().getContextClassLoader()).newInstance();
        } catch (Exception e) {
            throw new FlinkHiveException("Unable to instantiate the hadoop input format", e);
        }
        ReflectionUtils.setConf(format, jobConf);
        jobConf.set(INPUT_DIR, sd.getLocation());
        //TODO: we should consider how to calculate the splits according to minNumSplits in the future.
        org.apache.hadoop.mapred.InputSplit[] splitArray = format.getSplits(jobConf, minNumSplits);
        for (org.apache.hadoop.mapred.InputSplit inputSplit : splitArray) {
            Preconditions.checkState(inputSplit instanceof FileSplit,
                    "Unsupported InputSplit type: " + inputSplit.getClass().getName());
            hiveSplits.add(new HiveSourceSplit((FileSplit) inputSplit, partition, null));
        }
    }

    return hiveSplits;
}

int lowerBound = logRunningTime("getNumFiles", numFiles);
if (lowerBound >= inferMaxParallelism) {
    parallelism = inferMaxParallelism;
    return this;
}

int splitNum = logRunningTime("createInputSplits", numSplits);
parallelism = Math.min(splitNum, inferMaxParallelism);

/**
 * Apply limit to calculate the parallelism.
 * Here limit is the limit in query <code>SELECT * FROM xxx LIMIT [limit]</code>.
 */
int limit(Long limit) {
    if (limit != null) {
        parallelism = Math.min(parallelism, (int) (limit / 1000));
    }

    // make sure that parallelism is at least 1
    return Math.max(1, parallelism);
}