@Override
  public Optional<ConnectorOutputMetadata> finishRefreshMaterializedView(
      ConnectorSession session,
      ConnectorTableHandle tableHandle,
      ConnectorInsertTableHandle insertHandle,
      Collection<Slice> fragments,
      Collection<ComputedStatistics> computedStatistics,
      List<ConnectorTableHandle> sourceTableHandles) {
    IcebergWritableTableHandle table = (IcebergWritableTableHandle) insertHandle;

    Table icebergTable = transaction.table();
    // delete before insert .. simulating overwrite
    transaction.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit();

    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);

    AppendFiles appendFiles = transaction.newFastAppend();
    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());
    }

    String dependencies =
        sourceTableHandles.stream()
            .map(
                handle -> {
                  if (!(handle instanceof IcebergTableHandle icebergHandle)) {
                    return UNKNOWN_SNAPSHOT_TOKEN;
                  }
                  return icebergHandle.getSchemaTableName()
                      + "="
                      + icebergHandle.getSnapshotId().map(Object.class::cast).orElse("");
                })
            .distinct()
            .collect(joining(","));

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

    // Update the 'dependsOnTables' property that tracks tables on which the materialized view
    // depends
    // and the corresponding snapshot ids of the tables
    appendFiles.set(DEPENDS_ON_TABLES, dependencies);
    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