@Override
  protected List<StatisticsFile> apply(Transaction transaction) {
    PartitionSpec spec = transaction.table().spec();
    if (this.addFiles.isEmpty()) {
      return Collections.emptyList();
    }

    Preconditions.checkNotNull(this.optimizedSequence, "optimized sequence must be set.");
    Preconditions.checkArgument(this.optimizedSequence > 0, "optimized sequence must > 0.");

    ReplacePartitions replacePartitions = transaction.newReplacePartitions();
    replacePartitions.set(MixedTableUtil.BLOB_TYPE_OPTIMIZED_SEQUENCE_EXIST, "true");
    addFiles.forEach(replacePartitions::addFile);
    replacePartitions.commit();
    CreateSnapshotEvent newSnapshot = (CreateSnapshotEvent) replacePartitions.updateEvent();

    StructLikeMap<Long> oldOptimizedSequence = MixedTableUtil.readOptimizedSequence(keyedTable);
    StructLikeMap<Long> optimizedSequence = StructLikeMap.create(spec.partitionType());
    if (oldOptimizedSequence != null) {
      optimizedSequence.putAll(oldOptimizedSequence);
    }
    addFiles.forEach(f -> optimizedSequence.put(f.partition(), this.optimizedSequence));

    Table table = transaction.table();
    StatisticsFile statisticsFile =
        StatisticsFileUtil.writerBuilder(table)
            .withSnapshotId(newSnapshot.snapshotId())
            .build()
            .add(
                MixedTableUtil.BLOB_TYPE_OPTIMIZED_SEQUENCE,
                optimizedSequence,
                StatisticsFileUtil.createPartitionDataSerializer(table.spec(), Long.class))
            .complete();
    return Collections.singletonList(statisticsFile);
  }
Replace {@link BaseTable} partition files and change max transaction id map
amoro-format-iceberg/src/main/java/org/apache/amoro/op/KeyedPartitionRewrite.java