public static long adjustSplitSize(long scanSize, int parallelism, long splitSize) {
    // use the configured split size if it produces at least one split per slot
    // otherwise, adjust the split size to target parallelism with a reasonable minimum
    // increasing the split size may cause expensive spills and is not done automatically
    long splitCount = LongMath.divide(scanSize, splitSize, RoundingMode.CEILING);
    long adjustedSplitSize = Math.max(scanSize / parallelism, Math.min(MIN_SPLIT_SIZE, splitSize));
    return splitCount < parallelism ? adjustedSplitSize : splitSize;
  }
@Test
  public void testAdaptiveSplitSize() {
    long scanSize = 500L * 1024 * 1024 * 1024; // 500 GB
    int parallelism = 500;
    long smallDefaultSplitSize = 128 * 1024 * 1024; // 128 MB
    long largeDefaultSplitSize = 2L * 1024 * 1024 * 1024; // 2 GB

    long adjusted1 = TableScanUtil.adjustSplitSize(scanSize, parallelism, smallDefaultSplitSize);
    assertThat(adjusted1).isEqualTo(smallDefaultSplitSize);

    long adjusted2 = TableScanUtil.adjustSplitSize(scanSize, parallelism, largeDefaultSplitSize);
    assertThat(adjusted2).isEqualTo(scanSize / parallelism);
  }