public static FileRewriteCoordinator get() {
    return INSTANCE;
  }
@Test
  public void testBinPackRewrite() throws NoSuchTableException, IOException {
    sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);

    Dataset<Row> df = newDF(1000);
    df.coalesce(1).writeTo(tableName).append();
    df.coalesce(1).writeTo(tableName).append();
    df.coalesce(1).writeTo(tableName).append();
    df.coalesce(1).writeTo(tableName).append();

    Table table = validationCatalog.loadTable(tableIdent);
    Assert.assertEquals("Should produce 4 snapshots", 4, Iterables.size(table.snapshots()));

    Dataset<Row> fileDF =
        spark.read().format("iceberg").load(tableName(tableIdent.name() + ".files"));
    List<Long> fileSizes = fileDF.select("file_size_in_bytes").as(Encoders.LONG()).collectAsList();
    long avgFileSize = fileSizes.stream().mapToLong(i -> i).sum() / fileSizes.size();

    try (CloseableIterable<FileScanTask> fileScanTasks = table.newScan().planFiles()) {
      String fileSetID = UUID.randomUUID().toString();

      ScanTaskSetManager taskSetManager = ScanTaskSetManager.get();
      taskSetManager.stageTasks(table, fileSetID, Lists.newArrayList(fileScanTasks));

      // read and pack original 4 files into 2 splits
      Dataset<Row> scanDF =
          spark
              .read()
              .format("iceberg")
              .option(SparkReadOptions.SCAN_TASK_SET_ID, fileSetID)
              .option(SparkReadOptions.SPLIT_SIZE, Long.toString(avgFileSize * 2))
              .option(SparkReadOptions.FILE_OPEN_COST, "0")
              .load(tableName);

      // write the packed data into new files where each split becomes a new file
      scanDF
          .writeTo(tableName)
          .option(SparkWriteOptions.REWRITTEN_FILE_SCAN_TASK_SET_ID, fileSetID)
          .append();

      // commit the rewrite
      FileRewriteCoordinator rewriteCoordinator = FileRewriteCoordinator.get();
      Set<DataFile> rewrittenFiles =
          taskSetManager.fetchTasks(table, fileSetID).stream()
              .map(t -> t.asFileScanTask().file())
              .collect(Collectors.toSet());
      Set<DataFile> addedFiles = rewriteCoordinator.fetchNewFiles(table, fileSetID);
      table.newRewrite().rewriteFiles(rewrittenFiles, addedFiles).commit();
    }

    table.refresh();

    Map<String, String> summary = table.currentSnapshot().summary();
    Assert.assertEquals("Deleted files count must match", "4", summary.get("deleted-data-files"));
    Assert.assertEquals("Added files count must match", "2", summary.get("added-data-files"));

    Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
    Assert.assertEquals("Row count must match", 4000L, rowCount);
  }