private void finishOptimize(
      ConnectorSession session,
      IcebergTableExecuteHandle executeHandle,
      Collection<Slice> fragments,
      List<Object> splitSourceInfo) {
    IcebergOptimizeHandle optimizeHandle =
        (IcebergOptimizeHandle) executeHandle.getProcedureHandle();
    Table icebergTable = transaction.table();

    // files to be deleted
    ImmutableSet.Builder<DataFile> scannedDataFilesBuilder = ImmutableSet.builder();
    ImmutableSet.Builder<DeleteFile> scannedDeleteFilesBuilder = ImmutableSet.builder();
    splitSourceInfo.stream()
        .map(DataFileWithDeleteFiles.class::cast)
        .forEach(
            dataFileWithDeleteFiles -> {
              scannedDataFilesBuilder.add(dataFileWithDeleteFiles.getDataFile());
              scannedDeleteFilesBuilder.addAll(dataFileWithDeleteFiles.getDeleteFiles());
            });

    Set<DataFile> scannedDataFiles = scannedDataFilesBuilder.build();
    Set<DeleteFile> fullyAppliedDeleteFiles = scannedDeleteFilesBuilder.build();

    List<CommitTaskData> commitTasks =
        fragments.stream()
            .map(slice -> commitTaskCodec.fromJson(slice.getBytes()))
            .collect(toImmutableList());

    Type[] partitionColumnTypes =
        icebergTable.spec().fields().stream()
            .map(
                field ->
                    field
                        .transform()
                        .getResultType(icebergTable.schema().findType(field.sourceId())))
            .toArray(Type[]::new);

    Set<DataFile> newFiles = new HashSet<>();
    for (CommitTaskData task : commitTasks) {
      DataFiles.Builder builder =
          DataFiles.builder(icebergTable.spec())
              .withPath(task.getPath())
              .withFileSizeInBytes(task.getFileSizeInBytes())
              .withFormat(optimizeHandle.getFileFormat().toIceberg())
              .withMetrics(task.getMetrics().metrics());

      if (!icebergTable.spec().fields().isEmpty()) {
        String partitionDataJson =
            task.getPartitionDataJson()
                .orElseThrow(() -> new VerifyException("No partition data for partitioned table"));
        builder.withPartition(PartitionData.fromJson(partitionDataJson, partitionColumnTypes));
      }

      newFiles.add(builder.build());
    }

    if (optimizeHandle.getSnapshotId().isEmpty()
        || scannedDataFiles.isEmpty() && fullyAppliedDeleteFiles.isEmpty() && newFiles.isEmpty()) {
      // Either the table is empty, or the table scan turned out to be empty, nothing to commit
      transaction = null;
      return;
    }

    // try to leave as little garbage as possible behind
    if (optimizeHandle.isRetriesEnabled()) {
      cleanExtraOutputFiles(
          session,
          newFiles.stream().map(dataFile -> dataFile.path().toString()).collect(toImmutableSet()));
    }

    RewriteFiles rewriteFiles = transaction.newRewrite();
    rewriteFiles.rewriteFiles(
        scannedDataFiles, fullyAppliedDeleteFiles, newFiles, ImmutableSet.of());
    // Table.snapshot method returns null if there is no matching snapshot
    Snapshot snapshot =
        requireNonNull(
            icebergTable.snapshot(optimizeHandle.getSnapshotId().get()), "snapshot is null");
    rewriteFiles.validateFromSnapshot(snapshot.snapshotId());
    commit(rewriteFiles, session);
    transaction.commitTransaction();
    transaction = null;
  }
Iceberg original metadata has some problems for mixed-format table, such as iceberg version,
table type. So copy from IcebergMetadata and made some change
amoro-format-mixed/amoro-mixed-trino/src/main/java/org/apache/amoro/trino/unkeyed/IcebergMetadata.java