Core: Add REST API for committing changes against multiple tables

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

@@ -107,6 +107,10 @@ public String tableName() {
   }
 
   public TableMetadata startMetadata() {
+    return base;
+  }
+
+  public TableMetadata currentMetadata() {
     return current;
   }
 

core/src/main/java/org/apache/iceberg/catalog/TableCommit.java

@@ -0,0 +1,52 @@
+/*
+ * 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.catalog;
+
+import java.util.List;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.UpdateRequirements;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.immutables.value.Value;
+
+@Value.Immutable
+public interface TableCommit {
+  TableIdentifier identifier();
+
+  List<UpdateRequirement> requirements();
+
+  List<MetadataUpdate> updates();
+
+  static TableCommit create(TableIdentifier identifier, TableMetadata base, TableMetadata updated) {
+    Preconditions.checkArgument(null != identifier, "Invalid table identifier: null");
+    Preconditions.checkArgument(null != base && null != updated, "Invalid table metadata: null");
+    Preconditions.checkArgument(
+        base.uuid().equals(updated.uuid()),
+        "UUID of base (%s) and updated (%s) table metadata does not match",
+        base.uuid(),
+        updated.uuid());
+
+    return ImmutableTableCommit.builder()
+        .identifier(identifier)
+        .requirements(UpdateRequirements.forUpdateTable(base, updated.changes()))
+        .updates(updated.changes())
+        .build();
+  }
+}

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

@@ -314,7 +314,7 @@ private static TableMetadata create(TableOperations ops, UpdateTableRequest requ
     return ops.current();
   }
 
-  private static TableMetadata commit(TableOperations ops, UpdateTableRequest request) {
+  static TableMetadata commit(TableOperations ops, UpdateTableRequest request) {
     AtomicBoolean isRetry = new AtomicBoolean(false);
     try {
       Tasks.foreach(ops)

core/src/main/java/org/apache/iceberg/rest/RESTCatalog.java

@@ -33,16 +33,19 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SessionCatalog;
 import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableCommit;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.hadoop.Configurable;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 
 public class RESTCatalog implements Catalog, SupportsNamespaces, Configurable<Object>, Closeable {
   private final RESTSessionCatalog sessionCatalog;
   private final Catalog delegate;
   private final SupportsNamespaces nsDelegate;
+  private final SessionCatalog.SessionContext context;
 
   public RESTCatalog() {
     this(
@@ -60,6 +63,7 @@ public RESTCatalog(
     this.sessionCatalog = new RESTSessionCatalog(clientBuilder, null);
     this.delegate = sessionCatalog.asCatalog(context);
     this.nsDelegate = (SupportsNamespaces) delegate;
+    this.context = context;
   }
 
   @Override
@@ -248,4 +252,13 @@ public void setConf(Object conf) {
   public void close() throws IOException {
     sessionCatalog.close();
   }
+
+  public void commitTransaction(List<TableCommit> commits) {
+    sessionCatalog.commitTransaction(context, commits);
+  }
+
+  public void commitTransaction(TableCommit... commits) {
+    sessionCatalog.commitTransaction(
+        context, ImmutableList.<TableCommit>builder().add(commits).build());
+  }
 }

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -52,6 +52,7 @@
 import org.apache.iceberg.catalog.BaseSessionCatalog;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableCommit;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
@@ -63,13 +64,16 @@
 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.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.rest.auth.OAuth2Properties;
 import org.apache.iceberg.rest.auth.OAuth2Util;
 import org.apache.iceberg.rest.auth.OAuth2Util.AuthSession;
+import org.apache.iceberg.rest.requests.CommitTransactionRequest;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.ConfigResponse;
 import org.apache.iceberg.rest.responses.CreateNamespaceResponse;
 import org.apache.iceberg.rest.responses.GetNamespaceResponse;
@@ -916,4 +920,20 @@ private Cache<TableOperations, FileIO> newFileIOCloser() {
                 })
         .build();
   }
+
+  public void commitTransaction(SessionContext context, List<TableCommit> commits) {
+    List<UpdateTableRequest> tableChanges = Lists.newArrayListWithCapacity(commits.size());
+
+    for (TableCommit commit : commits) {
+      tableChanges.add(
+          new UpdateTableRequest(commit.identifier(), commit.requirements(), commit.updates()));
+    }
+
+    client.post(
+        paths.commitTransaction(),
+        new CommitTransactionRequest(tableChanges),
+        null,
+        headers(context),
+        ErrorHandlers.tableCommitHandler());
+  }
 }

core/src/main/java/org/apache/iceberg/rest/requests/UpdateTableRequest.java

@@ -48,7 +48,7 @@ public UpdateTableRequest(
     this.updates = updates;
   }
 
-  UpdateTableRequest(
+  public UpdateTableRequest(
       TableIdentifier identifier,
       List<org.apache.iceberg.UpdateRequirement> requirements,
       List<MetadataUpdate> updates) {

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -22,6 +22,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.BaseTransaction;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.Transactions;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -40,6 +45,8 @@
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.base.Splitter;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.rest.requests.CommitTransactionRequest;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
@@ -130,7 +137,9 @@ enum Route {
         HTTPMethod.POST,
         "v1/namespaces/{namespace}/tables/{table}/metrics",
         ReportMetricsRequest.class,
-        null);
+        null),
+    COMMIT_TRANSACTION(
+        HTTPMethod.POST, "v1/transactions/commit", CommitTransactionRequest.class, null);
 
     private final HTTPMethod method;
     private final int requiredLength;
@@ -357,12 +366,49 @@ public <T extends RESTResponse> T handleRequest(
           return null;
         }
 
+      case COMMIT_TRANSACTION:
+        {
+          CommitTransactionRequest request = castRequest(CommitTransactionRequest.class, body);
+          commitTransaction(catalog, request);
+          return null;
+        }
+
       default:
     }
 
     return null;
   }
 
+  /**
+   * This is a very simplistic approach that only validates the requirements for each table and does
+   * not do any other conflict detection. Therefore, it does not guarantee true transactional
+   * atomicity, which is left to the implementation details of a REST server.
+   */
+  private static void commitTransaction(Catalog catalog, CommitTransactionRequest request) {
+    List<Transaction> transactions = Lists.newArrayList();
+
+    for (UpdateTableRequest tableChange : request.tableChanges()) {
+      Table table = catalog.loadTable(tableChange.identifier());
+      if (table instanceof BaseTable) {
+        Transaction transaction =
+            Transactions.newTransaction(
+                tableChange.identifier().toString(), ((BaseTable) table).operations());
+        transactions.add(transaction);
+
+        BaseTransaction.TransactionTable txTable =
+            (BaseTransaction.TransactionTable) transaction.table();
+
+        // this performs validations and makes temporary commits that are in-memory
+        CatalogHandlers.commit(txTable.operations(), tableChange);
+      } else {
+        throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable");
+      }
+    }
+
+    // only commit if validations passed previously
+    transactions.forEach(Transaction::commitTransaction);
+  }
+
   public <T extends RESTResponse> T execute(
       HTTPMethod method,
       String path,