static int inferParallelism(
      ReadableConfig readableConfig, long limitCount, Supplier<Integer> splitCountProvider) {
    int parallelism =
        readableConfig.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
    if (readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
      int maxInferParallelism =
          readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
      Preconditions.checkState(
          maxInferParallelism >= 1,
          FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key()
              + " cannot be less than 1");
      parallelism = Math.min(splitCountProvider.get(), maxInferParallelism);
    }

    if (limitCount > 0) {
      int limit = limitCount >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) limitCount;
      parallelism = Math.min(parallelism, limit);
    }

    // parallelism must be positive.
    parallelism = Math.max(1, parallelism);
    return parallelism;
  }
@Test
  public void testInferedParallelism() throws IOException {
    Configuration configuration = new Configuration();
    // Empty table, infer parallelism should be at least 1
    int parallelism = SourceUtil.inferParallelism(configuration, -1L, () -> 0);
    assertThat(parallelism).isEqualTo(1);

    // 2 splits (max infer is the default value 100 , max > splits num), the parallelism is splits
    // num : 2
    parallelism = SourceUtil.inferParallelism(configuration, -1L, () -> 2);
    assertThat(parallelism).isEqualTo(2);

    // 2 splits and limit is 1 , max infer parallelism is default 100，
    // which is greater than splits num and limit, the parallelism is the limit value : 1
    parallelism = SourceUtil.inferParallelism(configuration, 1, () -> 2);
    assertThat(parallelism).isEqualTo(1);

    // 2 splits and max infer parallelism is 1 (max < splits num), the parallelism is  1
    configuration.setInteger(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX, 1);
    parallelism = SourceUtil.inferParallelism(configuration, -1L, () -> 2);
    assertThat(parallelism).isEqualTo(1);

    // 2 splits, max infer parallelism is 1, limit is 3, the parallelism is max infer parallelism :
    // 1
    parallelism = SourceUtil.inferParallelism(configuration, 3, () -> 2);
    assertThat(parallelism).isEqualTo(1);

    // 2 splits, infer parallelism is disabled, the parallelism is flink default parallelism 1
    configuration.setBoolean(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
    parallelism = SourceUtil.inferParallelism(configuration, 3, () -> 2);
    assertThat(parallelism).isEqualTo(1);
  }