@Override
  public Optional<ConnectorOutputMetadata> finishInsert(
      ConnectorSession session,
      ConnectorInsertTableHandle insertHandle,
      Collection<Slice> fragments,
      Collection<ComputedStatistics> computedStatistics) {
    List<CommitTaskData> commitTasks =
        fragments.stream()
            .map(slice -> commitTaskCodec.fromJson(slice.getBytes()))
            .collect(toImmutableList());

    if (commitTasks.isEmpty()) {
      transaction = null;
      return Optional.empty();
    }

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

    AppendFiles appendFiles = transaction.newAppend();
    ImmutableSet.Builder<String> writtenFiles = ImmutableSet.builder();
    for (CommitTaskData task : commitTasks) {
      DataFiles.Builder builder =
          DataFiles.builder(icebergTable.spec())
              .withPath(task.getPath())
              .withFileSizeInBytes(task.getFileSizeInBytes())
              .withFormat(table.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));
      }

      appendFiles.appendFile(builder.build());
      writtenFiles.add(task.getPath());
    }

    // try to leave as little garbage as possible behind
    if (table.getRetryMode() != NO_RETRIES) {
      cleanExtraOutputFiles(session, writtenFiles.build());
    }

    commit(appendFiles, session);
    transaction.commitTransaction();
    transaction = null;

    return Optional.of(
        new HiveWrittenPartitions(
            commitTasks.stream().map(CommitTaskData::getPath).collect(toImmutableList())));
  }
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