Core, Hive: Fix NPE in commit status check when a create commit fails

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

@@ -333,6 +333,13 @@ protected CommitStatus checkCommitStatusStrict(String newMetadataLocation, Table
    */
   private boolean checkCurrentMetadataLocation(String newMetadataLocation) {
     TableMetadata metadata = refresh();
+    if (metadata == null) {
+      // the table does not exist in the catalog. this happens when a create-table commit fails
+      // before the table is persisted, in which case the new metadata location cannot be the
+      // current one or part of the table's history
+      return false;
+    }
+
     String currentMetadataFileLocation = metadata.metadataFileLocation();
     return currentMetadataFileLocation.equals(newMetadataLocation)
         || metadata.previousFiles().stream()

core/src/test/java/org/apache/iceberg/TestBaseMetastoreTableOperations.java

@@ -0,0 +1,104 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.apache.iceberg.BaseMetastoreOperations.CommitStatus;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+public class TestBaseMetastoreTableOperations {
+
+  private static final Schema SCHEMA =
+      new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
+
+  private static final Map<String, String> FAST_STATUS_CHECKS =
+      ImmutableMap.of(
+          TableProperties.COMMIT_NUM_STATUS_CHECKS, "1",
+          TableProperties.COMMIT_STATUS_CHECKS_MIN_WAIT_MS, "1",
+          TableProperties.COMMIT_STATUS_CHECKS_MAX_WAIT_MS, "10",
+          TableProperties.COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS, "100");
+
+  /**
+   * Mimics metastore-backed table operations for a table that was never persisted to the catalog,
+   * e.g. when a CREATE TABLE commit fails before the table is stored in the metastore. Like {@code
+   * HiveTableOperations#doRefresh()}, a missing table is not an error when no metadata location is
+   * known, and refreshing from a null metadata location leaves the current metadata null.
+   */
+  private static class NeverPersistedTableOperations extends BaseMetastoreTableOperations {
+
+    @Override
+    protected String tableName() {
+      return "db.never_persisted";
+    }
+
+    @Override
+    public FileIO io() {
+      return null;
+    }
+
+    @Override
+    protected void doRefresh() {
+      refreshFromMetadataLocation(null, 1);
+    }
+
+    private CommitStatus strictStatus(String newMetadataLocation, TableMetadata config) {
+      return checkCommitStatusStrict(newMetadataLocation, config);
+    }
+
+    private CommitStatus status(String newMetadataLocation, TableMetadata config) {
+      return checkCommitStatus(newMetadataLocation, config);
+    }
+  }
+
+  @Test
+  public void strictStatusCheckIsFailureWhenTableWasNeverPersisted() {
+    NeverPersistedTableOperations ops = new NeverPersistedTableOperations();
+    TableMetadata metadata =
+        TableMetadata.newTableMetadata(
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            "file:/tmp/db/never_persisted",
+            FAST_STATUS_CHECKS);
+
+    assertThat(
+            ops.strictStatus(
+                "file:/tmp/db/never_persisted/metadata/00000-uuid.metadata.json", metadata))
+        .isEqualTo(CommitStatus.FAILURE);
+  }
+
+  @Test
+  public void statusCheckIsUnknownWhenTableWasNeverPersisted() {
+    NeverPersistedTableOperations ops = new NeverPersistedTableOperations();
+    TableMetadata metadata =
+        TableMetadata.newTableMetadata(
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            "file:/tmp/db/never_persisted",
+            FAST_STATUS_CHECKS);
+
+    assertThat(
+            ops.status("file:/tmp/db/never_persisted/metadata/00000-uuid.metadata.json", metadata))
+        .isEqualTo(CommitStatus.UNKNOWN);
+  }
+}

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java

@@ -276,6 +276,13 @@ public void doCommit(ViewMetadata base, ViewMetadata metadata) {
    */
   private boolean checkCurrentMetadataLocation(String newMetadataLocation) {
     ViewMetadata metadata = refresh();
+    if (metadata == null) {
+      // the view does not exist in the catalog. this happens when a create-view commit fails
+      // before the view is persisted, in which case the new metadata location cannot be the
+      // current one
+      return false;
+    }
+
     return newMetadataLocation.equals(metadata.metadataFileLocation());
   }
 

hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveViewCommits.java

@@ -41,6 +41,7 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.CommitFailedException;
@@ -50,6 +51,7 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.view.BaseView;
 import org.apache.iceberg.view.ImmutableSQLViewRepresentation;
+import org.apache.iceberg.view.ImmutableViewVersion;
 import org.apache.iceberg.view.View;
 import org.apache.iceberg.view.ViewMetadata;
 import org.apache.thrift.TException;
@@ -216,6 +218,78 @@ public void testThriftExceptionUnknownStateIfNotInHistoryFailureOnCommit()
         .isEqualTo(2);
   }
 
+  /**
+   * Pretends we throw an unclear error while persisting a create-view commit, for a view that was
+   * never stored in the metastore. The commit status check must resolve cleanly instead of NPE-ing:
+   * the view-specific {@code doRefresh} treats a missing view as non-fatal when no metadata
+   * location is known and refreshes from a null location, so the status-check supplier observes
+   * null current metadata and resolves to false (a new metadata location cannot be current for a
+   * view that does not exist). The relaxed check then maps that to UNKNOWN.
+   */
+  @Test
+  public void testThriftExceptionUnknownStateOnCreateCommitWhenViewNeverPersisted()
+      throws TException, InterruptedException, IOException {
+    TableIdentifier createIdentifier = TableIdentifier.of(NS, "create_commit_failed_view");
+    HiveViewOperations ops = (HiveViewOperations) catalog.newViewOps(createIdentifier);
+    HiveViewOperations spyOps = spy(ops);
+
+    failCommitAndThrowException(spyOps);
+
+    Path createLocation = new Path(viewLocation.getParent(), "create_commit_failed_view");
+    ViewMetadata metadata =
+        ViewMetadata.builder()
+            .setLocation(createLocation.toString())
+            .setProperties(
+                ImmutableMap.of(
+                    TableProperties.COMMIT_NUM_STATUS_CHECKS, "1",
+                    TableProperties.COMMIT_STATUS_CHECKS_MIN_WAIT_MS, "1",
+                    TableProperties.COMMIT_STATUS_CHECKS_MAX_WAIT_MS, "10",
+                    TableProperties.COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS, "100"))
+            .setCurrentVersion(
+                ImmutableViewVersion.builder()
+                    .versionId(1)
+                    .schemaId(SCHEMA.schemaId())
+                    .timestampMillis(System.currentTimeMillis())
+                    .defaultNamespace(NS)
+                    .putSummary("operation", "create")
+                    .addRepresentations(
+                        ImmutableSQLViewRepresentation.builder()
+                            .sql(VIEW_QUERY)
+                            .dialect("hive")
+                            .build())
+                    .build(),
+                SCHEMA)
+            .build();
+
+    try {
+      assertThatThrownBy(() -> spyOps.commit(null, metadata))
+          .isInstanceOf(CommitStateUnknownException.class)
+          .hasMessageStartingWith("Datacenter on fire");
+
+      assertThat(catalog.viewExists(createIdentifier))
+          .as("The view should not have been created")
+          .isFalse();
+
+      // pins the view-specific doRefresh wiring: a missing view is not an error when no metadata
+      // location is known, so refreshing a never-persisted view must yield null metadata
+      assertThat(ops.refresh())
+          .as("Refreshing a never-persisted view should yield null metadata")
+          .isNull();
+
+      // and the commit status check supplier must resolve to false for the null metadata instead
+      // of throwing an NPE
+      assertThat(
+              checkCurrentMetadataLocation(
+                  ops, createLocation + "/metadata/00000-uuid.metadata.json"))
+          .as("A new metadata location cannot be current for a never-persisted view")
+          .isFalse();
+    } finally {
+      createLocation
+          .getFileSystem(HIVE_METASTORE_EXTENSION.hiveConf())
+          .delete(createLocation, true);
+    }
+  }
+
   /** Pretends we throw an error while persisting that actually does commit serverside. */
   @Test
   public void testThriftExceptionSuccessOnCommit() throws TException, InterruptedException {
@@ -594,6 +668,20 @@ private void breakFallbackCatalogCommitCheck(HiveViewOperations spyOperations) {
         .thenThrow(new RuntimeException("Still on fire")); // Failure on commit check
   }
 
+  private static boolean checkCurrentMetadataLocation(
+      HiveViewOperations ops, String newMetadataLocation) {
+    try {
+      return (Boolean)
+          ReflectionSupport.invokeMethod(
+              HiveViewOperations.class.getDeclaredMethod(
+                  "checkCurrentMetadataLocation", String.class),
+              ops,
+              newMetadataLocation);
+    } catch (NoSuchMethodException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   private boolean metadataFileExists(ViewMetadata metadata) {
     return new File(metadata.metadataFileLocation().replace("file:", "")).exists();
   }