public static <T extends ScanTask> List<ScanTaskGroup<T>> planTaskGroups(
      List<T> tasks, long splitSize, int lookback, long openFileCost) {
    return Lists.newArrayList(
        planTaskGroups(CloseableIterable.withNoopClose(tasks), splitSize, lookback, openFileCost));
  }
@Test
  public void testTaskGroupPlanningByPartition() {
    // When all files belong to the same partition, we should combine them together as long as the
    // total file size is <= split size
    List<PartitionScanTask> tasks =
        ImmutableList.of(
            taskWithPartition(SPEC1, PARTITION1, 64),
            taskWithPartition(SPEC1, PARTITION1, 128),
            taskWithPartition(SPEC1, PARTITION1, 64),
            taskWithPartition(SPEC1, PARTITION1, 128));

    int count = 0;
    for (ScanTaskGroup<PartitionScanTask> task :
        TableScanUtil.planTaskGroups(tasks, 512, 10, 4, SPEC1.partitionType())) {
      assertThat(task.filesCount()).isEqualTo(4);
      assertThat(task.sizeBytes()).isEqualTo(64 + 128 + 64 + 128);
      count += 1;
    }
    assertThat(count).isOne();

    // We have 2 files from partition 1 and 2 files from partition 2, so they should be combined
    // separately
    tasks =
        ImmutableList.of(
            taskWithPartition(SPEC1, PARTITION1, 64),
            taskWithPartition(SPEC1, PARTITION1, 128),
            taskWithPartition(SPEC1, PARTITION2, 64),
            taskWithPartition(SPEC1, PARTITION2, 128));

    count = 0;
    for (ScanTaskGroup<PartitionScanTask> task :
        TableScanUtil.planTaskGroups(tasks, 512, 10, 4, SPEC1.partitionType())) {
      assertThat(task.filesCount()).isEqualTo(2);
      assertThat(task.sizeBytes()).isEqualTo(64 + 128);
      count += 1;
    }
    assertThat(count).isEqualTo(2);

    // Similar to the case above, but now files have different partition specs
    tasks =
        ImmutableList.of(
            taskWithPartition(SPEC1, PARTITION1, 64),
            taskWithPartition(SPEC2, PARTITION1, 128),
            taskWithPartition(SPEC1, PARTITION2, 64),
            taskWithPartition(SPEC2, PARTITION2, 128));

    count = 0;
    for (ScanTaskGroup<PartitionScanTask> task :
        TableScanUtil.planTaskGroups(tasks, 512, 10, 4, SPEC1.partitionType())) {
      assertThat(task.filesCount()).isEqualTo(2);
      assertThat(task.sizeBytes()).isEqualTo(64 + 128);
      count += 1;
    }
    assertThat(count).isEqualTo(2);

    // Combining within partitions should also respect split size. In this case, the split size
    // is equal to the file size, so each partition will have 2 tasks.
    tasks =
        ImmutableList.of(
            taskWithPartition(SPEC1, PARTITION1, 128),
            taskWithPartition(SPEC2, PARTITION1, 128),
            taskWithPartition(SPEC1, PARTITION2, 128),
            taskWithPartition(SPEC2, PARTITION2, 128));

    count = 0;
    for (ScanTaskGroup<PartitionScanTask> task :
        TableScanUtil.planTaskGroups(tasks, 128, 10, 4, SPEC1.partitionType())) {
      assertThat(task.filesCount()).isOne();
      assertThat(task.sizeBytes()).isEqualTo(128);
      count += 1;
    }
    assertThat(count).isEqualTo(4);

    // The following should throw exception since `SPEC2` is not an intersection of partition specs
    // across all tasks.
    List<PartitionScanTask> tasks2 =
        ImmutableList.of(
            taskWithPartition(SPEC1, PARTITION1, 128), taskWithPartition(SPEC2, PARTITION2, 128));

    assertThatThrownBy(
            () -> TableScanUtil.planTaskGroups(tasks2, 128, 10, 4, SPEC2.partitionType()))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageStartingWith("Cannot find field");
  }