[No Merge] Changes with BaseMetadata

aws/src/main/java/org/apache/iceberg/aws/dynamodb/DynamoDbTableOperations.java

@@ -23,6 +23,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
+import org.apache.iceberg.BaseMetadata;
 import org.apache.iceberg.BaseMetastoreTableOperations;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.aws.AwsProperties;
@@ -102,7 +103,9 @@ protected void doRefresh() {
   }
 
   @Override
-  protected void doCommit(TableMetadata base, TableMetadata metadata) {
+  protected void doCommit(BaseMetadata baseMetadata, BaseMetadata newMetadata) {
+    TableMetadata metadata = (TableMetadata) newMetadata;
+    TableMetadata base = (TableMetadata) baseMetadata;
     boolean newTable = base == null;
     String newMetadataLocation = writeNewMetadataIfRequired(newTable, metadata);
     CommitStatus commitStatus = CommitStatus.FAILURE;

aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java

@@ -21,6 +21,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
+import org.apache.iceberg.BaseMetadata;
 import org.apache.iceberg.BaseMetastoreTableOperations;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
@@ -142,7 +143,10 @@ protected void doRefresh() {
   }
 
   @Override
-  protected void doCommit(TableMetadata base, TableMetadata metadata) {
+  protected void doCommit(BaseMetadata baseMetadata, BaseMetadata newMetadata) {
+    TableMetadata metadata = (TableMetadata) newMetadata;
+    TableMetadata base = (TableMetadata) baseMetadata;
+
     CommitStatus commitStatus = CommitStatus.FAILURE;
     RetryDetector retryDetector = new RetryDetector();
 

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

@@ -0,0 +1,53 @@
+/*
+ * 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.io.Serializable;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.iceberg.util.PropertyUtil;
+
+/** A base class for {@link TableMetadata} and {@link org.apache.iceberg.view.ViewMetadata} */
+public interface BaseMetadata extends Serializable {
+
+  String location();
+
+  Map<String, String> properties();
+
+  @Nullable
+  String metadataFileLocation();
+
+  Schema schema();
+
+  default String property(String property, String defaultValue) {
+    return properties().getOrDefault(property, defaultValue);
+  }
+
+  default boolean propertyAsBoolean(String property, boolean defaultValue) {
+    return PropertyUtil.propertyAsBoolean(properties(), property, defaultValue);
+  }
+
+  default int propertyAsInt(String property, int defaultValue) {
+    return PropertyUtil.propertyAsInt(properties(), property, defaultValue);
+  }
+
+  default long propertyAsLong(String property, long defaultValue) {
+    return PropertyUtil.propertyAsLong(properties(), property, defaultValue);
+  }
+}

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

@@ -0,0 +1,161 @@
+/*
+ * 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 static org.apache.iceberg.TableProperties.COMMIT_NUM_STATUS_CHECKS;
+import static org.apache.iceberg.TableProperties.COMMIT_NUM_STATUS_CHECKS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MAX_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MAX_WAIT_MS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MIN_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MIN_WAIT_MS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS_DEFAULT;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.Tasks;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BaseMetastoreOperations {
+  private static final Logger LOG = LoggerFactory.getLogger(BaseMetastoreOperations.class);
+
+  public void commitHelper(BaseMetadata base, BaseMetadata metadata) {
+    String op = metadata instanceof TableMetadata ? "Table" : "View";
+    String opLowerCase = op.toLowerCase();
+    String opName = opName();
+    // if the metadata is already out of date, reject it
+    if (isStale(base)) {
+      if (base != null) {
+        throw new CommitFailedException("Cannot commit: stale table metadata");
+      } else {
+        // when current is non-null, the table exists. but when base is null, the commit is trying
+        // to create the table
+        throw new AlreadyExistsException("Table already exists: %s", opName());
+      }
+    }
+    // if the metadata is not changed, return early
+    if (base == metadata) {
+      LOG.info("Nothing to commit.");
+      return;
+    }
+
+    long start = System.currentTimeMillis();
+    doCommit(base, metadata);
+    deleteRemovedMetadataFiles(base, metadata);
+    requestRefresh();
+
+    LOG.info(
+        "Successfully committed to {} {} in {} ms",
+        opLowerCase,
+        opName,
+        System.currentTimeMillis() - start);
+  }
+
+  protected abstract String opName();
+
+  protected abstract boolean isStale(BaseMetadata baseMetadata);
+
+  protected abstract void requestRefresh();
+
+  protected abstract void deleteRemovedMetadataFiles(
+      BaseMetadata baseMetadata, BaseMetadata newMetadata);;
+
+  protected void doCommit(BaseMetadata baseMetadata, BaseMetadata newMetadata) {
+    throw new UnsupportedOperationException("Not implemented: doCommit");
+  }
+
+  /**
+   * Attempt to load the table and see if any current or past metadata location matches the one we
+   * were attempting to set. This is used as a last resort when we are dealing with exceptions that
+   * may indicate the commit has failed but don't have proof that this is the case. Note that all
+   * the previous locations must also be searched on the chance that a second committer was able to
+   * successfully commit on top of our commit.
+   *
+   * @param tableName full name of the table
+   * @param newMetadataLocation the path of the new commit file
+   * @param properties properties for retry
+   * @param loadMetadataLocations supply all the metadata locations
+   * @return Commit Status of Success, Failure or Unknown
+   */
+  protected BaseMetastoreTableOperations.CommitStatus checkCommitStatus(
+      String tableName,
+      String newMetadataLocation,
+      Map<String, String> properties,
+      Supplier<List<String>> loadMetadataLocations) {
+    int maxAttempts =
+        PropertyUtil.propertyAsInt(
+            properties, COMMIT_NUM_STATUS_CHECKS, COMMIT_NUM_STATUS_CHECKS_DEFAULT);
+    long minWaitMs =
+        PropertyUtil.propertyAsLong(
+            properties, COMMIT_STATUS_CHECKS_MIN_WAIT_MS, COMMIT_STATUS_CHECKS_MIN_WAIT_MS_DEFAULT);
+    long maxWaitMs =
+        PropertyUtil.propertyAsLong(
+            properties, COMMIT_STATUS_CHECKS_MAX_WAIT_MS, COMMIT_STATUS_CHECKS_MAX_WAIT_MS_DEFAULT);
+    long totalRetryMs =
+        PropertyUtil.propertyAsLong(
+            properties,
+            COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS,
+            COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS_DEFAULT);
+
+    AtomicReference<BaseMetastoreTableOperations.CommitStatus> status =
+        new AtomicReference<>(BaseMetastoreTableOperations.CommitStatus.UNKNOWN);
+
+    Tasks.foreach(newMetadataLocation)
+        .retry(maxAttempts)
+        .suppressFailureWhenFinished()
+        .exponentialBackoff(minWaitMs, maxWaitMs, totalRetryMs, 2.0)
+        .onFailure(
+            (location, checkException) ->
+                LOG.error("Cannot check if commit to {} exists.", tableName, checkException))
+        .run(
+            location -> {
+              List<String> allMetadataLocations = loadMetadataLocations.get();
+              boolean commitSuccess = allMetadataLocations.contains(newMetadataLocation);
+
+              if (commitSuccess) {
+                LOG.info(
+                    "Commit status check: Commit to {} of {} succeeded",
+                    tableName,
+                    newMetadataLocation);
+                status.set(BaseMetastoreTableOperations.CommitStatus.SUCCESS);
+              } else {
+                LOG.warn(
+                    "Commit status check: Commit to {} of {} unknown, new metadata location is not current "
+                        + "or in history",
+                    tableName,
+                    newMetadataLocation);
+              }
+            });
+
+    if (status.get() == BaseMetastoreTableOperations.CommitStatus.UNKNOWN) {
+      LOG.error(
+          "Cannot determine commit state to {}. Failed during checking {} times. "
+              + "Treating commit state as unknown.",
+          tableName,
+          maxAttempts);
+    }
+    return status.get();
+  }
+}