Core: Throw CommitStateUnknownException if RuntimeException that is n…

…ot marked as cleanable is thrown

core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java

@@ -31,6 +31,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.exceptions.CleanableFailure;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
@@ -308,12 +309,15 @@ private void replaceDataFiles(
       LOG.warn("Commit state unknown, cannot clean up files that may have been committed", e);
       throw e;
     } catch (Exception e) {
-      LOG.warn("Failed to commit rewrite, cleaning up rewritten files", e);
-      Tasks.foreach(Iterables.transform(addedDataFiles, f -> f.path().toString()))
-          .noRetry()
-          .suppressFailureWhenFinished()
-          .onFailure((location, exc) -> LOG.warn("Failed to delete: {}", location, exc))
-          .run(fileIO::deleteFile);
+      if (e instanceof CleanableFailure) {
+        LOG.warn("Failed to commit rewrite, cleaning up rewritten files", e);
+        Tasks.foreach(Iterables.transform(addedDataFiles, f -> f.path().toString()))
+            .noRetry()
+            .suppressFailureWhenFinished()
+            .onFailure((location, exc) -> LOG.warn("Failed to delete: {}", location, exc))
+            .run(fileIO::deleteFile);
+      }
+
       throw e;
     }
   }

core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java

@@ -23,6 +23,7 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.RewriteFiles;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.CleanableFailure;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -119,8 +120,12 @@ public void commitOrClean(Set<RewriteFileGroup> rewriteGroups) {
           e);
       throw e;
     } catch (Exception e) {
-      LOG.error("Cannot commit groups {}, attempting to clean up written files", rewriteGroups, e);
-      rewriteGroups.forEach(this::abortFileGroup);
+      if (e instanceof CleanableFailure) {
+        LOG.error(
+            "Cannot commit groups {}, attempting to clean up written files", rewriteGroups, e);
+        rewriteGroups.forEach(this::abortFileGroup);
+      }
+
       throw e;
     }
   }

core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java

@@ -24,6 +24,7 @@
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.RewriteFiles;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.CleanableFailure;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -102,8 +103,12 @@ public void commitOrClean(Set<RewritePositionDeletesGroup> rewriteGroups) {
           e);
       throw e;
     } catch (Exception e) {
-      LOG.error("Cannot commit groups {}, attempting to clean up written files", rewriteGroups, e);
-      rewriteGroups.forEach(this::abort);
+      if (e instanceof CleanableFailure) {
+        LOG.error(
+            "Cannot commit groups {}, attempting to clean up written files", rewriteGroups, e);
+        rewriteGroups.forEach(this::abort);
+      }
+
       throw e;
     }
   }

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java

@@ -46,6 +46,7 @@
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.actions.ImmutableRewriteManifests;
 import org.apache.iceberg.actions.RewriteManifests;
+import org.apache.iceberg.exceptions.CleanableFailure;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.io.OutputFile;
@@ -354,10 +355,13 @@ private void replaceManifests(
     } catch (CommitStateUnknownException commitStateUnknownException) {
       // don't clean up added manifest files, because they may have been successfully committed.
       throw commitStateUnknownException;
-    } catch (Exception e) {
-      // delete all new manifests because the rewrite failed
-      deleteFiles(Iterables.transform(addedManifests, ManifestFile::path));
-      throw e;
+    } catch (Throwable failure) {
+      if (failure instanceof CleanableFailure) {
+        // delete all new manifests because the rewrite failed
+        deleteFiles(Iterables.transform(addedManifests, ManifestFile::path));
+      }
+
+      throw failure;
     }
   }
 

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java

@@ -44,7 +44,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.deletes.DeleteGranularity;
 import org.apache.iceberg.deletes.PositionDelete;
-import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.exceptions.CleanableFailure;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.BasePositionDeltaWriter;
@@ -105,7 +105,7 @@ class SparkPositionDeltaWrite implements DeltaWrite, RequiresDistributionAndOrde
   private final Context context;
   private final Map<String, String> writeProperties;
 
-  private boolean cleanupOnAbort = true;
+  private boolean cleanupOnAbort = false;
 
   SparkPositionDeltaWrite(
       SparkSession spark,
@@ -304,9 +304,12 @@ private void commitOperation(SnapshotUpdate<?> operation, String description) {
         operation.commit(); // abort is automatically called if this fails
         long duration = System.currentTimeMillis() - start;
         LOG.info("Committed in {} ms", duration);
-      } catch (CommitStateUnknownException commitStateUnknownException) {
-        cleanupOnAbort = false;
-        throw commitStateUnknownException;
+      } catch (Throwable failure) {
+        if (failure instanceof CleanableFailure) {
+          cleanupOnAbort = true;
+        }
+
+        throw failure;
       }
     }
   }

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java

@@ -40,7 +40,7 @@
 import org.apache.iceberg.SnapshotSummary;
 import org.apache.iceberg.SnapshotUpdate;
 import org.apache.iceberg.Table;
-import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.exceptions.CleanableFailure;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.ClusteredDataWriter;
@@ -102,7 +102,7 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
   private final SparkWriteRequirements writeRequirements;
   private final Map<String, String> writeProperties;
 
-  private boolean cleanupOnAbort = true;
+  private boolean cleanupOnAbort = false;
 
   SparkWrite(
       SparkSession spark,
@@ -233,9 +233,12 @@ private void commitOperation(SnapshotUpdate<?> operation, String description) {
       operation.commit(); // abort is automatically called if this fails
       long duration = System.currentTimeMillis() - start;
       LOG.info("Committed in {} ms", duration);
-    } catch (CommitStateUnknownException commitStateUnknownException) {
-      cleanupOnAbort = false;
-      throw commitStateUnknownException;
+    } catch (Throwable failure) {
+      if (failure instanceof CleanableFailure) {
+        cleanupOnAbort = true;
+      }
+
+      throw failure;
     }
   }