@Override
  public Row[] execute() throws AnalysisException {
    List<DataFile> dataFiles;
    TableIdentifier ident;
    PartitionSpec spec;
    Schema schema;
    LOG.info(
        "start to migrate {} to {}, using temp backup table {}",
        sourceIdentifier,
        targetIdentifier,
        backupV1TableIdentifier);
    V1Table sourceTable = loadV1Table(sourceCatalog, backupV1TableIdentifier);
    ident =
        new TableIdentifier(
            backupV1TableIdentifier.name(), Some.apply(backupV1TableIdentifier.namespace()[0]));
    dataFiles = loadDataFiles(ident);
    UnkeyedTable table = createUnkeyedTable(sourceTable);

    spec = table.spec();

    AppendFiles appendFiles = table.newAppend();
    dataFiles.forEach(appendFiles::appendFile);
    appendFiles.commit();

    LOG.info(
        "migrate table {} finished, remove metadata of backup {} table",
        targetIdentifier,
        backupV1TableIdentifier);

    if (PartitionSpec.unpartitioned().equals(spec)) {
      return new Row[] {RowFactory.create("ALL", dataFiles.size())};
    }

    Map<String, List<DataFile>> partitions = Maps.newHashMap();
    dataFiles.forEach(
        d -> {
          String partition = spec.partitionToPath(d.partition());
          List<DataFile> df = partitions.computeIfAbsent(partition, p -> Lists.newArrayList());
          df.add(d);
        });
    return partitions.keySet().stream()
        .sorted()
        .map(p -> RowFactory.create(p, partitions.get(p).size()))
        .toArray(Row[]::new);
  }
migrate a v1 table to mixed-format table. will reuse file in v1 table , but delete metadata in
session catalog
amoro-format-mixed/amoro-mixed-spark/v3.2/amoro-mixed-spark-3.2/src/main/java/org/apache/amoro/spark/command/MigrateToMixedFormatCommand.java