Support setting table statistics

Implement `Transaction.updateStatistics` API.

api/src/main/java/org/apache/iceberg/UpdateStatistics.java

@@ -18,8 +18,10 @@
  */
 package org.apache.iceberg;
 
+import java.util.List;
+
 /** API for updating statistics files in a table. */
-public interface UpdateStatistics extends PendingUpdate<Snapshot> {
+public interface UpdateStatistics extends PendingUpdate<List<StatisticsFile>> {
   /**
    * Set the table's statistics file for given snapshot, replacing the previous statistics file for
    * the snapshot if any exists.

core/src/main/java/org/apache/iceberg/BaseTransaction.java

@@ -216,6 +216,14 @@ public DeleteFiles newDelete() {
     return delete;
   }
 
+  @Override
+  public UpdateStatistics updateStatistics() {
+    checkLastOperationCommitted("UpdateStatistics");
+    UpdateStatistics updateStatistics = new SetStatistics(transactionOps);
+    updates.add(updateStatistics);
+    return updateStatistics;
+  }
+
   @Override
   public ExpireSnapshots expireSnapshots() {
     checkLastOperationCommitted("ExpireSnapshots");

core/src/main/java/org/apache/iceberg/SetStatistics.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Stream;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+public class SetStatistics implements UpdateStatistics {
+  private final TableOperations ops;
+  private final Map<Long, Optional<StatisticsFile>> statisticsToSet = Maps.newHashMap();
+
+  public SetStatistics(TableOperations ops) {
+    this.ops = ops;
+  }
+
+  @Override
+  public UpdateStatistics setStatistics(long snapshotId, StatisticsFile statisticsFile) {
+    Preconditions.checkArgument(snapshotId == statisticsFile.snapshotId());
+    statisticsToSet.put(snapshotId, Optional.of(statisticsFile));
+    return this;
+  }
+
+  @Override
+  public UpdateStatistics removeStatistics(long snapshotId) {
+    statisticsToSet.put(snapshotId, Optional.empty());
+    return this;
+  }
+
+  @Override
+  public List<StatisticsFile> apply() {
+    return Stream.concat(
+            // Retained statistics
+            ops.current().statisticsFiles().stream()
+                .filter(
+                    statisticsFile -> !statisticsToSet.containsKey(statisticsFile.snapshotId())),
+            // New statistics
+            statisticsToSet.values().stream().filter(Optional::isPresent).map(Optional::get))
+        .collect(ImmutableList.toImmutableList());
+  }
+
+  @Override
+  public void commit() {
+    TableMetadata base = ops.current(); // or ops.refresh() ?
+    TableMetadata.Builder builder = TableMetadata.buildFrom(base);
+    statisticsToSet.forEach(
+        (snapshotId, statistics) -> {
+          if (statistics.isPresent()) {
+            builder.setStatistics(snapshotId, statistics.get());
+          } else {
+            builder.removeStatistics(snapshotId);
+          }
+        });
+    TableMetadata newMetadata = builder.build();
+    ops.commit(base, newMetadata);
+  }
+}