Avoid deleting table metadata on ambiguous commit failures

CHANGELOG.md

@@ -28,6 +28,9 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ## [Unreleased]
 
 ### Highlights
+- Table and view commits no longer delete newly written metadata when the metastore write outcome
+  is unknown (e.g. a dropped connection); they still clean up on known failures. JDBC reports these
+  errors as `PersistenceCommitStateUnknownException` (HTTP 500).
 
 ### Upgrade notes
 

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperations.java

@@ -60,6 +60,14 @@ public class DatasourceOperations {
   private static final String UNIQUENESS_CONSTRAINT_VIOLATION_SQL_CODE = "23505";
   private static final String RELATION_DOES_NOT_EXIST = "42P01";
 
+  // SQLSTATE codes treated as ambiguous commit outcomes (the write may already have committed).
+  // Class 08 (connection exception) is SQL-standard and portable across databases.
+  private static final String CONNECTION_EXCEPTION_SQL_STATE_CLASS = "08";
+  // POSTGRES: query_canceled, e.g. statement_timeout (also CockroachDB via PG compatibility)
+  private static final String POSTGRES_QUERY_CANCELED_SQL_STATE = "57014";
+  // ODBC: timeout expired (used by some drivers, e.g. jTDS and older MySQL connectors)
+  private static final String ODBC_TIMEOUT_SQL_STATE = "HYT00";
+
   // H2 STATUS CODES
   // 90079 = Schema not found, 42S02 = Table or view not found
   private static final String H2_SCHEMA_DOES_NOT_EXIST = "90079";
@@ -178,7 +186,8 @@ public <T> void executeSelectOverStream(
             executeSelectOverStreamWithConnection(query, converterInstance, consumer, connection);
             return null;
           }
-        });
+        },
+        false);
   }
 
   /** Connection-aware version for use inside runWithinTransaction. */
@@ -192,7 +201,8 @@ public <T> void executeSelectOverStream(
         () -> {
           executeSelectOverStreamWithConnection(query, converterInstance, consumer, connection);
           return null;
-        });
+        },
+        false);
   }
 
   /**
@@ -358,7 +368,52 @@ public Integer execute(Connection connection, QueryGenerator.PreparedQuery prepa
     }
   }
 
-  private boolean isRetryable(SQLException e) {
+  /**
+   * Whether a SQLException indicates the statement may already have been applied on the server
+   * while the client cannot confirm the outcome (connection loss, timeout, cancellation).
+   *
+   * <p>Callers must not treat these as definite failures for cleanup or safe retry of CAS writes.
+   */
+  public boolean isAmbiguousCommitOutcome(SQLException e) {
+    if (e == null) {
+      return false;
+    }
+    if (e instanceof java.sql.SQLTimeoutException) {
+      return true;
+    }
+    if (e instanceof java.sql.SQLTransientConnectionException
+        || e instanceof java.sql.SQLNonTransientConnectionException) {
+      return true;
+    }
+    String sqlState = e.getSQLState();
+    if (sqlState != null) {
+      if (sqlState.startsWith(CONNECTION_EXCEPTION_SQL_STATE_CLASS)
+          || sqlState.equals(POSTGRES_QUERY_CANCELED_SQL_STATE)
+          || sqlState.equals(ODBC_TIMEOUT_SQL_STATE)) {
+        return true;
+      }
+    }
+    String message = e.getMessage();
+    if (message != null) {
+      String lower = message.toLowerCase(Locale.ROOT);
+      return lower.contains("connection reset")
+          || lower.contains("connection refused")
+          || lower.contains("connection is closed")
+          || lower.contains("broken pipe")
+          || lower.contains("query canceled")
+          || lower.contains("canceling statement due to statement timeout");
+    }
+    return false;
+  }
+
+  private boolean isRetryable(SQLException e, boolean mutating) {
+    // For mutating operations an ambiguous outcome (connection loss, timeout, cancellation) may
+    // mean the write already committed under auto-commit, so retrying risks a double-apply or a
+    // misreported result. Reads have no commit outcome to protect, so they stay retryable.
+    if (mutating && isAmbiguousCommitOutcome(e)) {
+      return false;
+    }
+
     String sqlState = e.getSQLState();
 
     if (sqlState != null) {
@@ -374,6 +429,13 @@ private boolean isRetryable(SQLException e) {
   // and more knobs for tuning retry pattern.
   @VisibleForTesting
   <T> T withRetries(Operation<T> operation) throws SQLException {
+    // Default to the mutating policy: it is the safe choice when the caller does not state whether
+    // the operation writes.
+    return withRetries(operation, true);
+  }
+
+  @VisibleForTesting
+  <T> T withRetries(Operation<T> operation, boolean mutating) throws SQLException {
     int attempts = 0;
     // maximum number of retries.
     int maxAttempts = relationalJdbcConfiguration.maxRetries().orElse(1);
@@ -409,7 +471,7 @@ <T> T withRetries(Operation<T> operation) throws SQLException {
         attempts++;
         long timeLeft =
             Math.max((maxRetryTime - TimeUnit.NANOSECONDS.toMillis(System.nanoTime())), 0L);
-        if (timeLeft == 0 || attempts >= maxAttempts || !isRetryable(sqlException)) {
+        if (timeLeft == 0 || attempts >= maxAttempts || !isRetryable(sqlException, mutating)) {
           String exceptionMessage =
               String.format(
                   "Failed due to '%s' (error code %d, sql-state '%s'), after %s attempts and %s milliseconds",

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -56,6 +56,7 @@
 import org.apache.polaris.core.persistence.BasePersistence;
 import org.apache.polaris.core.persistence.EntityAlreadyExistsException;
 import org.apache.polaris.core.persistence.IntegrationPersistence;
+import org.apache.polaris.core.persistence.PersistenceCommitStateUnknownException;
 import org.apache.polaris.core.persistence.PolicyMappingAlreadyExistsException;
 import org.apache.polaris.core.persistence.PrincipalSecretsGenerator;
 import org.apache.polaris.core.persistence.RetryOnConcurrencyException;
@@ -144,7 +145,7 @@ public void writeEntity(
             return datasourceOperations.executeUpdate(preparedQuery);
           });
     } catch (SQLException e) {
-      throw new RuntimeException("Error persisting entity", e);
+      throw wrapEntityWriteFailure(e);
     }
   }
 
@@ -169,10 +170,7 @@ public void writeEntities(
             return true;
           });
     } catch (SQLException e) {
-      throw new RuntimeException(
-          String.format(
-              "Error executing the transaction for writing entities due to %s", e.getMessage()),
-          e);
+      throw wrapEntityWriteFailure(e, "Error executing the transaction for writing entities");
     }
   }
 
@@ -236,8 +234,7 @@ private void persistEntity(
           throw new RetryOnConcurrencyException(
               e, "Conflicting entity is not visible in the current transaction snapshot; retry");
         }
-        throw new RuntimeException(
-            String.format("Failed to write entity due to %s", e.getMessage()), e);
+        throw wrapEntityWriteFailure(e);
       }
     } else {
       // CAS on both entity_version and grant_records_version because grant operations only
@@ -274,12 +271,23 @@ private void persistEntity(
               originalEntity.getGrantRecordsVersion());
         }
       } catch (SQLException e) {
-        throw new RuntimeException(
-            String.format("Failed to write entity due to %s", e.getMessage()), e);
+        throw wrapEntityWriteFailure(e);
       }
     }
   }
 
+  private RuntimeException wrapEntityWriteFailure(SQLException e) {
+    return wrapEntityWriteFailure(e, "Failed to write entity");
+  }
+
+  private RuntimeException wrapEntityWriteFailure(SQLException e, String context) {
+    if (datasourceOperations.isAmbiguousCommitOutcome(e)) {
+      return new PersistenceCommitStateUnknownException(
+          String.format("%s due to %s; commit outcome is unknown", context, e.getMessage()), e);
+    }
+    return new RuntimeException(String.format("%s due to %s", context, e.getMessage()), e);
+  }
+
   @Override
   public void writeToGrantRecords(
       @NonNull PolarisCallContext callCtx, @NonNull PolarisGrantRecord grantRec) {

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperationsTest.java

@@ -228,18 +228,19 @@ void testSuccessfulExecutionOnFirstAttempt() throws SQLException {
 
   @Test
   void testSuccessfulExecutionAfterOneRetry() throws SQLException {
-    when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(4));
+    when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(3));
     when(relationalJdbcConfiguration.maxDurationInMs()).thenReturn(Optional.of(2000L));
     when(relationalJdbcConfiguration.initialDelayInMs()).thenReturn(Optional.of(0L));
+    // Only serialization failures (definite rollback) are retryable; connection-class errors are
+    // treated as ambiguous commit outcomes and must not be retried.
     when(mockOperation.execute())
         .thenThrow(new SQLException("Retryable error", "40001"))
-        .thenThrow(new SQLException("connection refused"))
-        .thenThrow(new SQLException("connection reset"))
+        .thenThrow(new SQLException("Retryable error", "40001"))
         .thenReturn("Success!");
 
     String result = datasourceOperations.withRetries(mockOperation);
     assertEquals("Success!", result);
-    verify(mockOperation, times(4)).execute();
+    verify(mockOperation, times(3)).execute();
   }
 
   @Test
@@ -323,4 +324,48 @@ void testDefaultConfigurationValues() throws SQLException {
     assertThrows(SQLException.class, () -> datasourceOperations.withRetries(mockOperation));
     verify(mockOperation, times(1)).execute();
   }
+
+  @Test
+  void isAmbiguousCommitOutcome_classifiesConnectionAndTimeoutFailures() {
+    assertTrue(datasourceOperations.isAmbiguousCommitOutcome(new SQLException("reset", "08006")));
+    assertTrue(
+        datasourceOperations.isAmbiguousCommitOutcome(new SQLException("canceled", "57014")));
+    assertTrue(
+        datasourceOperations.isAmbiguousCommitOutcome(
+            new SQLException("Connection reset by peer")));
+    assertTrue(
+        datasourceOperations.isAmbiguousCommitOutcome(new java.sql.SQLTimeoutException("timeout")));
+    // Serialization failure is a definite rollback — retryable, not ambiguous commit success.
+    assertTrue(
+        !datasourceOperations.isAmbiguousCommitOutcome(new SQLException("serialization", "40001")));
+    // Constraint violations are definite failures.
+    assertTrue(!datasourceOperations.isAmbiguousCommitOutcome(new SQLException("unique", "23505")));
+  }
+
+  @Test
+  void withRetries_doesNotRetryAmbiguousConnectionFailure() throws Exception {
+    when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(3));
+    when(relationalJdbcConfiguration.maxDurationInMs()).thenReturn(Optional.of(5000L));
+    when(relationalJdbcConfiguration.initialDelayInMs()).thenReturn(Optional.of(1L));
+    when(mockOperation.execute()).thenThrow(new SQLException("Connection reset", "08006"));
+
+    assertThrows(SQLException.class, () -> datasourceOperations.withRetries(mockOperation));
+    verify(mockOperation, times(1)).execute();
+  }
+
+  @Test
+  void withRetries_retriesAmbiguousConnectionFailureForReads() throws Exception {
+    when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(3));
+    when(relationalJdbcConfiguration.maxDurationInMs()).thenReturn(Optional.of(5000L));
+    when(relationalJdbcConfiguration.initialDelayInMs()).thenReturn(Optional.of(1L));
+    // A read has no commit outcome to protect, so a transient connection failure is retried.
+    when(mockOperation.execute())
+        .thenThrow(new SQLException("Connection reset by peer"))
+        .thenThrow(new SQLException("Connection reset by peer"))
+        .thenReturn("Success!");
+
+    String result = datasourceOperations.withRetries(mockOperation, false);
+    assertEquals("Success!", result);
+    verify(mockOperation, times(3)).execute();
+  }
 }

polaris-core/src/main/java/org/apache/polaris/core/persistence/PersistenceCommitStateUnknownException.java

@@ -0,0 +1,51 @@
+/*
+ * 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.polaris.core.persistence;
+
+import jakarta.ws.rs.core.Response;
+import org.apache.polaris.core.exceptions.PolarisException;
+
+/**
+ * Thrown when a persistence write may have committed but the client cannot confirm the outcome (for
+ * example a connection drop or timeout after the database applied an auto-commit update). Named to
+ * distinguish it from Iceberg's table-commit {@code CommitStateUnknownException}.
+ *
+ * <p>Callers must not treat this as a definite failure: newly written metadata files must not be
+ * deleted, and clients should not blindly retry the same commit.
+ *
+ * <p>Note: the simple class name is surfaced to clients as the {@code type} field of the REST error
+ * payload (see {@code PolarisExceptionMapper}). Renaming this class changes that wire value, which
+ * clients may use to recognize an unknown commit outcome.
+ */
+public class PersistenceCommitStateUnknownException extends PolarisException {
+
+  public PersistenceCommitStateUnknownException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public PersistenceCommitStateUnknownException(String message) {
+    super(message);
+  }
+
+  @Override
+  public int httpStatusCode() {
+    // Iceberg REST: unknown commit outcome must be 5xx so clients do not treat it as cleanable.
+    return Response.Status.INTERNAL_SERVER_ERROR.getStatusCode();
+  }
+}