@Override
  public List<? extends BoundedSource<Row>> split(
      long desiredBundleSizeBytes, PipelineOptions options) throws Exception {
    ArrayList<ScanTaskSource> splits = new ArrayList<>();

    switch (scanConfig.getScanType()) {
      case TABLE:
        TableScan tableScan = getTableScan();
        if (desiredBundleSizeBytes > 0) {
          tableScan =
              tableScan.option(TableProperties.SPLIT_SIZE, Long.toString(desiredBundleSizeBytes));
        }

        try (CloseableIterable<CombinedScanTask> tasks = tableScan.planTasks()) {
          for (CombinedScanTask combinedScanTask : tasks) {
            splits.add(new ScanTaskSource(scanConfig, combinedScanTask));
          }
        } catch (IOException e) {
          throw new RuntimeException(e);
        }
        break;
      case BATCH:
        throw new UnsupportedOperationException("BATCH scan not supported");
      default:
        throw new UnsupportedOperationException("Unknown scan type: " + scanConfig.getScanType());
    }

    return splits;
  }
@Test
  public void testInitialSplitting() throws Exception {
    TableIdentifier tableId =
        TableIdentifier.of("default", "table" + Long.toString(UUID.randomUUID().hashCode(), 16));
    Table simpleTable = warehouse.createTable(tableId, TestFixtures.SCHEMA);
    simpleTable
        .newFastAppend()
        .appendFile(
            warehouse.writeRecords(
                "file1s1.parquet", simpleTable.schema(), TestFixtures.FILE1SNAPSHOT1))
        .appendFile(
            warehouse.writeRecords(
                "file2s1.parquet", simpleTable.schema(), TestFixtures.FILE2SNAPSHOT1))
        .appendFile(
            warehouse.writeRecords(
                "file3s1.parquet", simpleTable.schema(), TestFixtures.FILE3SNAPSHOT1))
        .commit();

    PipelineOptions options = PipelineOptionsFactory.create();

    Map<String, String> catalogProps =
        ImmutableMap.<String, String>builder()
            .put("type", CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP)
            .put("warehouse", warehouse.location)
            .build();

    BoundedSource<Row> source =
        new ScanSource(
            IcebergScanConfig.builder()
                .setCatalogConfig(
                    IcebergCatalogConfig.builder()
                        .setCatalogName("name")
                        .setCatalogProperties(catalogProps)
                        .build())
                .setScanType(IcebergScanConfig.ScanType.TABLE)
                .setTableIdentifier(simpleTable.name().replace("hadoop.", "").split("\\."))
                .setSchema(IcebergUtils.icebergSchemaToBeamSchema(TestFixtures.SCHEMA))
                .build());

    // Input data for this test is tiny so try a number of very small split sizes
    SourceTestUtils.assertSourcesEqualReferenceSource(source, source.split(1, options), options);
    SourceTestUtils.assertSourcesEqualReferenceSource(source, source.split(2, options), options);
    SourceTestUtils.assertSourcesEqualReferenceSource(source, source.split(5, options), options);
    SourceTestUtils.assertSourcesEqualReferenceSource(source, source.split(10, options), options);
    SourceTestUtils.assertSourcesEqualReferenceSource(source, source.split(100, options), options);
    SourceTestUtils.assertSourcesEqualReferenceSource(source, source.split(1000, options), options);
  }