public static FileRewriteCoordinator get() {
    return INSTANCE;
  }
@TestTemplate
  public void testSortRewrite() 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);
    assertThat(table.snapshots()).as("Should produce 4 snapshots").hasSize(4);

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

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

      // read original 4 files as 4 splits
      Dataset<Row> scanDF =
          spark
              .read()
              .format("iceberg")
              .option(SparkReadOptions.SCAN_TASK_SET_ID, fileSetID)
              .option(SparkReadOptions.SPLIT_SIZE, "134217728")
              .option(SparkReadOptions.FILE_OPEN_COST, "134217728")
              .load(tableName);

      // make sure we disable AQE and set the number of shuffle partitions as the target num files
      ImmutableMap<String, String> sqlConf =
          ImmutableMap.of(
              "spark.sql.shuffle.partitions", "2",
              "spark.sql.adaptive.enabled", "false");

      withSQLConf(
          sqlConf,
          () -> {
            try {
              // write new files with sorted records
              scanDF
                  .sort("id")
                  .writeTo(tableName)
                  .option(SparkWriteOptions.REWRITTEN_FILE_SCAN_TASK_SET_ID, fileSetID)
                  .append();
            } catch (NoSuchTableException e) {
              throw new RuntimeException("Could not replace files", e);
            }
          });

      // 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();
    assertThat(summary.get("deleted-data-files"))
        .as("Deleted files count must match")
        .isEqualTo("4");
    assertThat(summary.get("added-data-files")).as("Added files count must match").isEqualTo("2");

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