@Override
  public void finishStatisticsCollection(
      ConnectorSession session,
      ConnectorTableHandle tableHandle,
      Collection<ComputedStatistics> computedStatistics) {
    IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
    Table table = transaction.table();
    if (handle.getSnapshotId().isEmpty()) {
      // No snapshot, table is empty
      verify(
          computedStatistics.isEmpty(),
          "Unexpected computed statistics that cannot be attached to a snapshot because none exists: %s",
          computedStatistics);

      // TODO (https://github.com/trinodb/trino/issues/15397): remove support for Trino-specific
      // statistics properties
      // Drop all stats. Empty table needs none
      UpdateProperties updateProperties = transaction.updateProperties();
      table.properties().keySet().stream()
          .filter(key -> key.startsWith(TRINO_STATS_PREFIX))
          .forEach(updateProperties::remove);
      updateProperties.commit();

      transaction.commitTransaction();
      transaction = null;
      return;
    }
    long snapshotId = handle.getSnapshotId().orElseThrow();

    Map<String, Integer> columnNameToId =
        table.schema().columns().stream()
            .collect(
                toImmutableMap(
                    nestedField -> nestedField.name().toLowerCase(ENGLISH),
                    Types.NestedField::fieldId));
    Set<Integer> columnIds = ImmutableSet.copyOf(columnNameToId.values());

    // TODO (https://github.com/trinodb/trino/issues/15397): remove support for Trino-specific
    // statistics properties
    // Drop stats for obsolete columns
    UpdateProperties updateProperties = transaction.updateProperties();
    table.properties().keySet().stream()
        .filter(
            key -> {
              if (!key.startsWith(TRINO_STATS_PREFIX)) {
                return false;
              }
              Matcher matcher = TRINO_STATS_COLUMN_ID_PATTERN.matcher(key);
              if (!matcher.matches()) {
                return false;
              }
              return !columnIds.contains(Integer.parseInt(matcher.group("columnId")));
            })
        .forEach(updateProperties::remove);
    updateProperties.commit();

    ImmutableMap.Builder<Integer, CompactSketch> ndvSketches = ImmutableMap.builder();
    for (ComputedStatistics computedStatistic : computedStatistics) {
      verify(
          computedStatistic.getGroupingColumns().isEmpty()
              && computedStatistic.getGroupingValues().isEmpty(),
          "Unexpected grouping");
      verify(computedStatistic.getTableStatistics().isEmpty(), "Unexpected table statistics");
      for (Map.Entry<ColumnStatisticMetadata, Block> entry :
          computedStatistic.getColumnStatistics().entrySet()) {
        ColumnStatisticMetadata statisticMetadata = entry.getKey();
        if (statisticMetadata.getConnectorAggregationId().equals(NUMBER_OF_DISTINCT_VALUES_NAME)) {
          Integer columnId =
              verifyNotNull(
                  columnNameToId.get(statisticMetadata.getColumnName()),
                  "Column not found in table: [%s]",
                  statisticMetadata.getColumnName());
          CompactSketch sketch = DataSketchStateSerializer.deserialize(entry.getValue(), 0);
          ndvSketches.put(columnId, sketch);
        } else {
          throw new UnsupportedOperationException("Unsupported statistic: " + statisticMetadata);
        }
      }
    }

    StatisticsFile statisticsFile =
        tableStatisticsWriter.writeStatisticsFile(
            session, table, snapshotId, ndvSketches.buildOrThrow());
    transaction.updateStatistics().setStatistics(snapshotId, statisticsFile).commit();

    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