API,Core: Multi-Table transactions API and support for REST

Author: nastra

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

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg;
 
+import java.util.List;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.ValidationException;
 
@@ -172,4 +173,14 @@ default ManageSnapshots manageSnapshots() {
    * @throws CommitFailedException If the updates cannot be committed due to conflicts.
    */
   void commitTransaction();
+
+  /** Rolls back any pending changes. */
+  default void rollback() {
+    throw new UnsupportedOperationException("Rollback not supported");
+  }
+
+  /** Provides access to the pending changes that are about to be committed. */
+  default List<PendingUpdate> pendingUpdates() {
+    throw new UnsupportedOperationException("Pending Updates not supported");
+  }
 }

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

@@ -99,7 +99,7 @@ protected Schema tableSchema() {
     return schema;
   }
 
-  protected TableScanContext context() {
+  public TableScanContext context() {
     return context;
   }
 

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

@@ -43,6 +43,7 @@
 import org.apache.iceberg.metrics.MetricsReporter;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 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.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.PropertyUtil;
@@ -107,6 +108,10 @@ public String tableName() {
   }
 
   public TableMetadata startMetadata() {
+    return base;
+  }
+
+  public TableMetadata currentMetadata() {
     return current;
   }
 
@@ -426,10 +431,10 @@ private void commitSimpleTransaction() {
       throw e;
 
     } catch (PendingUpdateFailedException e) {
-      cleanUpOnCommitFailure();
+      rollback();
       throw e.wrapped();
     } catch (RuntimeException e) {
-      cleanUpOnCommitFailure();
+      rollback();
       throw e;
     }
 
@@ -468,7 +473,8 @@ private void commitSimpleTransaction() {
     }
   }
 
-  private void cleanUpOnCommitFailure() {
+  @Override
+  public void rollback() {
     // the commit failed and no files were committed. clean up each update.
     Tasks.foreach(updates)
         .suppressFailureWhenFinished()
@@ -486,6 +492,11 @@ private void cleanUpOnCommitFailure() {
         .run(ops.io()::deleteFile);
   }
 
+  @Override
+  public List<PendingUpdate> pendingUpdates() {
+    return ImmutableList.copyOf(updates);
+  }
+
   private void applyUpdates(TableOperations underlyingOps) {
     if (base != underlyingOps.refresh()) {
       // use refreshed the metadata

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

@@ -18,6 +18,8 @@
  */
 package org.apache.iceberg;
 
+import java.util.List;
+
 class CommitCallbackTransaction implements Transaction {
   static Transaction addCallback(Transaction txn, Runnable callback) {
     return new CommitCallbackTransaction(txn, callback);
@@ -121,4 +123,14 @@ public void commitTransaction() {
     wrapped.commitTransaction();
     callback.run();
   }
+
+  @Override
+  public void rollback() {
+    wrapped.rollback();
+  }
+
+  @Override
+  public List<PendingUpdate> pendingUpdates() {
+    return wrapped.pendingUpdates();
+  }
 }

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

@@ -28,8 +28,11 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.rest.RESTCatalogTransaction;
+import org.apache.iceberg.rest.RESTSessionCatalog;
 
 public abstract class BaseSessionCatalog implements SessionCatalog {
   private final Cache<String, Catalog> catalogs =
@@ -62,7 +65,13 @@ public <T> T withContext(SessionContext context, Function<Catalog, T> task) {
     return task.apply(asCatalog(context));
   }
 
-  public class AsCatalog implements Catalog, SupportsNamespaces {
+  public void commitTransaction(SessionContext context, List<TableCommit> commits) {
+    throw new UnsupportedOperationException(
+        "commitTransaction is not supported by catalog " + name());
+  }
+
+  public class AsCatalog implements Catalog, SupportsNamespaces, SupportsCatalogTransactions {
+
     private final SessionContext context;
 
     private AsCatalog(SessionContext context) {
@@ -159,5 +168,19 @@ public boolean removeProperties(Namespace namespace, Set<String> removals) {
     public boolean namespaceExists(Namespace namespace) {
       return BaseSessionCatalog.this.namespaceExists(context, namespace);
     }
+
+    @Override
+    public CatalogTransaction createTransaction(CatalogTransaction.IsolationLevel isolationLevel) {
+      Preconditions.checkState(
+          BaseSessionCatalog.this instanceof RESTSessionCatalog,
+          "Only RESTSessionCatalog currently supports CatalogTransactions");
+
+      return new RESTCatalogTransaction(
+          this, (RESTSessionCatalog) BaseSessionCatalog.this, context, isolationLevel);
+    }
+
+    public void commitTransaction(List<TableCommit> commits) {
+      BaseSessionCatalog.this.commitTransaction(context, commits);
+    }
   }
 }

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

@@ -0,0 +1,84 @@
+/*
+ * 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;
+
+public interface CatalogTransaction {
+
+  enum IsolationLevel {
+
+    /**
+     * All reads that are being made will see the last committed values that existed when the table
+     * was loaded first inside the catalog transaction. Subsequent changes to a table that happened
+     * outside the catalog transaction after the table was read will not be seen to prevent <b>read
+     * skew</b> (reading a table multiple times within the catalog transaction should always return
+     * the same results). <br>
+     * <br>
+     * Will successfully commit only if the values updated by the transaction do not conflict with
+     * other concurrent updates. <br>
+     * <br>
+     *
+     * <p>Note that under SNAPSHOT isolation a <b>write skew anomaly</b> is acceptable and
+     * permitted. In a <b>write skew anomaly</b>, two transactions (T1 and T2) concurrently read an
+     * overlapping data set (e.g. values V1 and V2), concurrently make disjoint updates (e.g. T1
+     * updates V1, T2 updates V2), and finally concurrently commit, neither having seen the update
+     * performed by the other.
+     */
+    SNAPSHOT,
+
+    /**
+     * All reads that are being made will see the last committed values that existed when the table
+     * was loaded first inside the catalog transaction. Subsequent changes to a table that happened
+     * outside the catalog transaction after the table was read will not be seen to prevent <b>read
+     * skew</b>.<br>
+     * <br>
+     * All tables participating in the transaction must be in the same state when committing
+     * compared to when the table was loaded first within the catalog transaction.<br>
+     * <br>
+     *
+     * <p>Note that a <b>write skew anomaly</b> is not possible under SERIALIZABLE isolation, where
+     * two transactions (T1 and T2) concurrently read an overlapping data set (e.g. values V1 and
+     * V2), concurrently make disjoint updates (e.g. T1 updates V1, T2 updates V2). This is because
+     * under SERIALIZABLE isolation either T1 or T2 would have to occur first and be visible to the
+     * other transaction.
+     */
+    SERIALIZABLE;
+  }
+
+  /**
+   * Performs an atomic commit of all the pending changes across multiple tables. Engine-specific
+   * implementations must ensure that all pending changes are applied atomically.
+   */
+  void commitTransaction();
+
+  /**
+   * Returns this catalog transaction as a {@link Catalog} API so that any actions that are called
+   * through this API are participating in this catalog transaction.
+   *
+   * @return This catalog transaction as a {@link Catalog} API. Any actions that are called through
+   *     this API are participating in this catalog transaction.
+   */
+  Catalog asCatalog();
+
+  /**
+   * Returns the current {@link IsolationLevel} for this transaction.
+   *
+   * @return The {@link IsolationLevel} for this transaction.
+   */
+  IsolationLevel isolationLevel();
+}

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

@@ -0,0 +1,32 @@
+/*
+ * 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 org.apache.iceberg.catalog.CatalogTransaction.IsolationLevel;
+
+public interface SupportsCatalogTransactions {
+
+  /**
+   * Create a new {@link CatalogTransaction} with the given {@link IsolationLevel}.
+   *
+   * @param isolationLevel The isolation level to use.
+   * @return A new {@link CatalogTransaction}.
+   */
+  CatalogTransaction createTransaction(IsolationLevel isolationLevel);
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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.immutables.value.Value;
+
+@Value.Immutable
+public interface TableCommit {
+  TableIdentifier identifier();
+
+  TableMetadata base();
+
+  List<MetadataUpdate> changes();
+}

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

@@ -40,6 +40,7 @@
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
 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;
@@ -49,8 +50,10 @@
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+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;
@@ -361,4 +364,37 @@ private static TableMetadata commit(TableOperations ops, UpdateTableRequest requ
 
     return ops.current();
   }
+
+  /**
+   * 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.
+   */
+  public static void commitTransaction(Catalog catalog, CommitTransactionRequest request) {
+    List<Transaction> transactions = Lists.newArrayList();
+
+    for (CommitTransactionRequest.CommitTableRequest tableChange : request.tableChanges()) {
+      Table table = catalog.loadTable(tableChange.identifier());
+      if (table instanceof BaseTable) {
+        UpdateTableRequest updateTableRequest =
+            new UpdateTableRequest(tableChange.requirements(), tableChange.updates());
+
+        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
+        commit(txTable.operations(), updateTableRequest);
+      } else {
+        throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable");
+      }
+    }
+
+    // only commit if validations passed previously
+    transactions.forEach(Transaction::commitTransaction);
+  }
 }