This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new edd5d6d19f [core] Preserve failures during manifest rewrite cleanup
(#9270)
edd5d6d19f is described below
commit edd5d6d19fb54ceeaf210700b0533c2bc2a234e4
Author: QuakeWang <[email protected]>
AuthorDate: Thu Aug 20 10:17:56 2026 +0800
[core] Preserve failures during manifest rewrite cleanup (#9270)
---
.../apache/paimon/manifest/ManifestAvroWriter.java | 108 ++-
.../operation/ManifestEntryExternalSort.java | 63 +-
.../operation/ManifestEntryRunMergePlan.java | 130 ++--
.../paimon/operation/ManifestFileBlockMerger.java | 2 +-
.../paimon/operation/ManifestFileLegacyMerger.java | 22 +-
.../paimon/operation/ManifestFileMerger.java | 13 +-
.../operation/ManifestRewriteCleanupTest.java | 778 +++++++++++++++++++++
7 files changed, 992 insertions(+), 124 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
index 16c4b68ab2..e78f273e29 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
@@ -33,7 +33,7 @@ import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.stats.SimpleStatsConverter;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.ObjectSerializer;
import org.apache.paimon.utils.PathFactory;
@@ -47,6 +47,8 @@ import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
/**
* Avro writer for manifest entries.
*
@@ -97,14 +99,19 @@ public final class ManifestAvroWriter implements
AutoCloseable {
currentWriter().write(entry);
afterWrite(1, false);
} catch (IOException | RuntimeException | Error failure) {
- abort();
+ abort(failure);
throw failure;
}
}
public void write(Iterable<? extends ManifestEntry> entries) throws
IOException {
- for (ManifestEntry entry : entries) {
- write(entry);
+ try {
+ for (ManifestEntry entry : entries) {
+ write(entry);
+ }
+ } catch (IOException | RuntimeException | Error failure) {
+ abort(failure);
+ throw failure;
}
}
@@ -113,7 +120,7 @@ public final class ManifestAvroWriter implements
AutoCloseable {
currentWriter().writeEncoded(encodedRecord, metadata);
afterWrite(1, false);
} catch (IOException | RuntimeException | Error failure) {
- abort();
+ abort(failure);
throw failure;
}
}
@@ -124,7 +131,7 @@ public final class ManifestAvroWriter implements
AutoCloseable {
currentWriter().writeRow(row, metadata);
afterWrite(1, false);
} catch (IOException | RuntimeException | Error failure) {
- abort();
+ abort(failure);
throw failure;
}
}
@@ -148,7 +155,7 @@ public final class ManifestAvroWriter implements
AutoCloseable {
currentWriter().writeEncodedBlock(block, metadata);
afterWrite(metadataRecordCount, true);
} catch (IOException | RuntimeException | Error failure) {
- abort();
+ abort(failure);
throw failure;
}
}
@@ -180,7 +187,7 @@ public final class ManifestAvroWriter implements
AutoCloseable {
fileWriter.collectStats(metadata);
afterWrite(copiedRecords, true);
} catch (IOException | RuntimeException | Error failure) {
- abort();
+ abort(failure);
throw failure;
}
}
@@ -228,14 +235,44 @@ public final class ManifestAvroWriter implements
AutoCloseable {
}
public void abort() {
+ Throwable cleanupFailure = abortCollecting(null);
+ if (cleanupFailure != null) {
+ ExceptionUtils.rethrow(cleanupFailure);
+ }
+ }
+
+ /** Aborts this writer and adds cleanup failures as suppressed exceptions.
*/
+ public void abort(Throwable primaryFailure) {
+ abortCollecting(checkNotNull(primaryFailure));
+ }
+
+ private Throwable abortCollecting(@Nullable Throwable primaryFailure) {
if (currentWriter != null) {
- currentWriter.abort();
+ primaryFailure = currentWriter.abortCollecting(primaryFailure,
true);
}
for (Path path : completedPaths) {
- fileIO.deleteQuietly(path);
+ try {
+ fileIO.deleteQuietly(path);
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
+ }
}
completedPaths.clear();
results.clear();
+ return primaryFailure;
+ }
+
+ private static Throwable closeCollecting(
+ @Nullable AutoCloseable closeable, @Nullable Throwable
primaryFailure) {
+ if (closeable == null) {
+ return primaryFailure;
+ }
+ try {
+ closeable.close();
+ } catch (Throwable cleanupFailure) {
+ primaryFailure = ExceptionUtils.firstOrSuppressed(cleanupFailure,
primaryFailure);
+ }
+ return primaryFailure;
}
@Override
@@ -246,7 +283,7 @@ public final class ManifestAvroWriter implements
AutoCloseable {
try {
closeCurrentWriter();
} catch (IOException | RuntimeException | Error failure) {
- abort();
+ abort(failure);
throw failure;
} finally {
closed = true;
@@ -365,6 +402,7 @@ public final class ManifestAvroWriter implements
AutoCloseable {
private boolean levelStatsKnown = true;
private @Nullable RowIdStats rowIdStats = new RowIdStats();
private boolean closed;
+ private boolean aborted;
private FileWriter(Path path) {
this.path = path;
@@ -378,19 +416,13 @@ public final class ManifestAvroWriter implements
AutoCloseable {
avroFileFormat.createBlockWriter(
out, ManifestEntry.MANIFEST_ROW_TYPE,
compression);
} catch (IOException failure) {
- IOUtils.closeQuietly(writer);
- IOUtils.closeQuietly(out);
- if (outputCreated) {
- fileIO.deleteQuietly(path);
- }
- throw new UncheckedIOException(
- "Failed to create manifest Avro writer for " + path,
failure);
+ UncheckedIOException primaryFailure =
+ new UncheckedIOException(
+ "Failed to create manifest Avro writer for " +
path, failure);
+ abortCollecting(primaryFailure, outputCreated);
+ throw primaryFailure;
} catch (RuntimeException | Error failure) {
- IOUtils.closeQuietly(writer);
- IOUtils.closeQuietly(out);
- if (outputCreated) {
- fileIO.deleteQuietly(path);
- }
+ abortCollecting(failure, outputCreated);
throw failure;
}
}
@@ -612,13 +644,31 @@ public final class ManifestAvroWriter implements
AutoCloseable {
}
}
- private void abort() {
- IOUtils.closeQuietly(writer);
+ private Throwable abortCollecting(@Nullable Throwable primaryFailure,
boolean deletePath) {
+ if (aborted) {
+ return primaryFailure;
+ }
+ aborted = true;
+ closed = true;
+ outputBytes = null;
+
+ AvroBlockWriter currentBlockWriter = writer;
writer = null;
- IOUtils.closeQuietly(out);
+ primaryFailure = closeCollecting(currentBlockWriter,
primaryFailure);
+
+ PositionOutputStream currentOut = out;
out = null;
- fileIO.deleteQuietly(path);
- closed = true;
+ primaryFailure = closeCollecting(currentOut, primaryFailure);
+
+ if (deletePath) {
+ try {
+ fileIO.deleteQuietly(path);
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
+ ExceptionUtils.firstOrSuppressed(cleanupFailure,
primaryFailure);
+ }
+ }
+ return primaryFailure;
}
private void close() throws IOException {
@@ -633,7 +683,7 @@ public final class ManifestAvroWriter implements
AutoCloseable {
out.close();
out = null;
} catch (IOException | RuntimeException | Error failure) {
- abort();
+ abortCollecting(failure, true);
throw failure;
} finally {
closed = true;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
index c261389093..00e8e38bcc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
@@ -35,6 +35,7 @@ import org.apache.paimon.manifest.ProjectedManifestEntry;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.MutableObjectIterator;
import org.apache.paimon.utils.Pair;
@@ -238,7 +239,8 @@ public class ManifestEntryExternalSort {
}
ManifestAvroWriter writer = manifestFile.createAvroWriter();
- Exception exception = null;
+ List<ManifestFileMeta> files = Collections.emptyList();
+ Throwable primaryFailure = null;
try {
MutableObjectIterator<BinaryRow> iterator =
sortBuffer.sortedIterator();
BinaryRow reuse = new
BinaryRow(sortKey.externalSortRowType().getFieldCount());
@@ -249,16 +251,19 @@ public class ManifestEntryExternalSort {
writer.write(entry.replace(sortKey.binaryManifestRow(row)));
}
entry.clear();
- } catch (Exception e) {
- exception = e;
+ writer.close();
+ files = writer.result();
+ } catch (Throwable failure) {
+ primaryFailure = failure;
} finally {
- if (exception != null) {
- writer.abort();
- throw exception;
+ if (primaryFailure != null) {
+ writer.abort(primaryFailure);
}
- writer.close();
}
- return writer.result();
+ if (primaryFailure != null) {
+ ExceptionUtils.rethrowException(primaryFailure);
+ }
+ return files;
}
private Pair<List<ManifestFileMeta>, List<ManifestFileMeta>>
writeMinorToManifest(
@@ -275,7 +280,9 @@ public class ManifestEntryExternalSort {
CompactFileIdentifierSet matchedEntries = new
CompactFileIdentifierSet();
CompactFileIdentifierSet emittedDeletes = new
CompactFileIdentifierSet();
ReusableIdentifier identifier = new ReusableIdentifier();
- Exception exception = null;
+ Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> files =
+ Pair.of(Collections.emptyList(), Collections.emptyList());
+ Throwable primaryFailure = null;
try {
MutableObjectIterator<BinaryRow> iterator =
sortBuffer.sortedIterator();
BinaryRow reuse = new
BinaryRow(sortKey.externalSortRowType().getFieldCount());
@@ -302,19 +309,37 @@ public class ManifestEntryExternalSort {
newFilesForAbort.addAll(addWriter.result());
deleteWriter.close();
newFilesForAbort.addAll(deleteWriter.result());
- } catch (Exception e) {
- exception = e;
+ files = Pair.of(addWriter.result(), deleteWriter.result());
+ } catch (Throwable failure) {
+ primaryFailure = failure;
} finally {
- identifier.release();
- matchedEntries.release();
- emittedDeletes.release();
- if (exception != null) {
- addWriter.abort();
- deleteWriter.abort();
- throw exception;
+ try {
+ identifier.release();
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
+ ExceptionUtils.firstOrSuppressed(cleanupFailure,
primaryFailure);
+ }
+ try {
+ matchedEntries.release();
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
+ ExceptionUtils.firstOrSuppressed(cleanupFailure,
primaryFailure);
+ }
+ try {
+ emittedDeletes.release();
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
+ ExceptionUtils.firstOrSuppressed(cleanupFailure,
primaryFailure);
+ }
+ if (primaryFailure != null) {
+ addWriter.abort(primaryFailure);
+ deleteWriter.abort(primaryFailure);
}
}
- return Pair.of(addWriter.result(), deleteWriter.result());
+ if (primaryFailure != null) {
+ ExceptionUtils.rethrowException(primaryFailure);
+ }
+ return files;
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
index 1b1d9562a0..618943df18 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
@@ -38,6 +38,7 @@ import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ProjectedManifestEntry;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.Pair;
import javax.annotation.Nullable;
@@ -71,7 +72,8 @@ final class ManifestEntryRunMergePlan {
List<ManifestFileMeta> mergeToManifest(
ManifestFile manifestFile, List<ManifestFileMeta>
newFilesForAbort) throws Exception {
List<Cursor> cursors = new ArrayList<>(sources.size());
- Exception failure = null;
+ List<ManifestFileMeta> files = Collections.emptyList();
+ Throwable primaryFailure = null;
try {
for (Source.Spec source : sources) {
Cursor cursor = source.open(manifestFile, deletes, minor,
partitions);
@@ -79,31 +81,31 @@ final class ManifestEntryRunMergePlan {
cursor.advance();
}
SelectionTree selectionTree = new SelectionTree(cursors);
- if (selectionTree.winner() < 0) {
- return Collections.emptyList();
- }
- List<ManifestFileMeta> files = writeSelected(selectionTree,
manifestFile);
- newFilesForAbort.addAll(files);
- return files;
- } catch (Exception e) {
- failure = e;
- throw e;
+ if (selectionTree.winner() >= 0) {
+ files = writeSelected(selectionTree, manifestFile);
+ newFilesForAbort.addAll(files);
+ }
+ } catch (Throwable failure) {
+ primaryFailure = failure;
} finally {
try {
closeCursors(cursors);
- } catch (Exception closeFailure) {
- if (failure == null) {
- throw closeFailure;
- }
- failure.addSuppressed(closeFailure);
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
}
}
+ if (primaryFailure != null) {
+ ExceptionUtils.rethrowException(primaryFailure);
+ }
+ return files;
}
Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> mergeMinorToManifest(
ManifestFile manifestFile, List<ManifestFileMeta>
newFilesForAbort) throws Exception {
List<Cursor> cursors = new ArrayList<>(sources.size());
- Exception failure = null;
+ Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> files =
+ Pair.of(Collections.emptyList(), Collections.emptyList());
+ Throwable primaryFailure = null;
try {
for (Source.Spec source : sources) {
Cursor cursor = source.open(manifestFile, deletes, minor,
partitions);
@@ -111,33 +113,31 @@ final class ManifestEntryRunMergePlan {
cursor.advance();
}
SelectionTree selectionTree = new SelectionTree(cursors);
- if (selectionTree.winner() < 0) {
- return Pair.of(Collections.emptyList(),
Collections.emptyList());
- }
- Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> files =
- writeMinorSelected(selectionTree, manifestFile, deletes);
- newFilesForAbort.addAll(files.getLeft());
- newFilesForAbort.addAll(files.getRight());
- return files;
- } catch (Exception e) {
- failure = e;
- throw e;
+ if (selectionTree.winner() >= 0) {
+ files = writeMinorSelected(selectionTree, manifestFile,
deletes);
+ newFilesForAbort.addAll(files.getLeft());
+ newFilesForAbort.addAll(files.getRight());
+ }
+ } catch (Throwable failure) {
+ primaryFailure = failure;
} finally {
try {
closeCursors(cursors);
- } catch (Exception closeFailure) {
- if (failure == null) {
- throw closeFailure;
- }
- failure.addSuppressed(closeFailure);
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
}
}
+ if (primaryFailure != null) {
+ ExceptionUtils.rethrowException(primaryFailure);
+ }
+ return files;
}
static List<ManifestFileMeta> writeSelected(
SelectionTree selectionTree, ManifestFile manifestFile) throws
Exception {
ManifestAvroWriter writer = manifestFile.createAvroWriter();
- Exception failure = null;
+ List<ManifestFileMeta> files = Collections.emptyList();
+ Throwable primaryFailure = null;
try {
int winner;
while ((winner = selectionTree.winner()) >= 0) {
@@ -152,16 +152,19 @@ final class ManifestEntryRunMergePlan {
writeCurrent(writer, cursor);
selectionTree.update(winner, cursor.advance());
}
- } catch (Exception e) {
- failure = e;
+ writer.close();
+ files = writer.result();
+ } catch (Throwable failure) {
+ primaryFailure = failure;
} finally {
- if (failure != null) {
- writer.abort();
- throw failure;
+ if (primaryFailure != null) {
+ writer.abort(primaryFailure);
}
- writer.close();
}
- return writer.result();
+ if (primaryFailure != null) {
+ ExceptionUtils.rethrowException(primaryFailure);
+ }
+ return files;
}
private static Pair<List<ManifestFileMeta>, List<ManifestFileMeta>>
writeMinorSelected(
@@ -171,7 +174,9 @@ final class ManifestEntryRunMergePlan {
ManifestAvroWriter deleteWriter = manifestFile.createAvroWriter();
CompactFileIdentifierSet matchedEntries = new
CompactFileIdentifierSet();
CompactFileIdentifierSet emittedDeletes = new
CompactFileIdentifierSet();
- Exception failure = null;
+ Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> files =
+ Pair.of(Collections.emptyList(), Collections.emptyList());
+ Throwable primaryFailure = null;
try {
int winner;
while ((winner = selectionTree.winner()) >= 0) {
@@ -203,18 +208,29 @@ final class ManifestEntryRunMergePlan {
}
addWriter.close();
deleteWriter.close();
- } catch (Exception e) {
- failure = e;
+ files = Pair.of(addWriter.result(), deleteWriter.result());
+ } catch (Throwable failure) {
+ primaryFailure = failure;
} finally {
- matchedEntries.release();
- emittedDeletes.release();
- if (failure != null) {
- addWriter.abort();
- deleteWriter.abort();
- throw failure;
+ try {
+ matchedEntries.release();
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
+ }
+ try {
+ emittedDeletes.release();
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
+ }
+ if (primaryFailure != null) {
+ addWriter.abort(primaryFailure);
+ deleteWriter.abort(primaryFailure);
}
}
- return Pair.of(addWriter.result(), deleteWriter.result());
+ if (primaryFailure != null) {
+ ExceptionUtils.rethrowException(primaryFailure);
+ }
+ return files;
}
private static void writeCurrent(ManifestAvroWriter writer, Cursor cursor)
throws Exception {
@@ -232,20 +248,16 @@ final class ManifestEntryRunMergePlan {
}
static void closeCursors(List<Cursor> cursors) throws Exception {
- Exception failure = null;
+ Throwable primaryFailure = null;
for (Cursor cursor : cursors) {
try {
cursor.close();
- } catch (Exception e) {
- if (failure == null) {
- failure = e;
- } else {
- failure.addSuppressed(e);
- }
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
}
}
- if (failure != null) {
- throw failure;
+ if (primaryFailure != null) {
+ ExceptionUtils.rethrowException(primaryFailure);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
index baef36d807..05b6225f60 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
@@ -483,7 +483,7 @@ final class ManifestFileBlockMerger {
writer.close();
return writer.result();
} catch (Exception | Error failure) {
- writer.abort();
+ writer.abort(failure);
throw failure;
} finally {
reusableIdentifier.release();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileLegacyMerger.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileLegacyMerger.java
index 12c6332526..a016fb0b55 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileLegacyMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileLegacyMerger.java
@@ -27,6 +27,7 @@ import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.Filter;
import org.slf4j.Logger;
@@ -218,7 +219,6 @@ final class ManifestFileLegacyMerger {
singletonList(
readForFullCompaction(
file, manifestFile, mustChange,
deleteEntries));
- Exception exception = null;
try {
for (FullCompactionReadResult readResult :
sequentialBatchedExecute(reader, toBeMerged,
manifestReadParallelism)) {
@@ -228,20 +228,16 @@ final class ManifestFileLegacyMerger {
result.add(readResult.file);
}
}
- } catch (Exception e) {
- exception = e;
- } finally {
- if (exception != null) {
- writer.abort();
- throw exception;
- }
writer.close();
+ List<ManifestFileMeta> merged = writer.result();
+ result.addAll(merged);
+ newFilesForAbort.addAll(merged);
+ return Optional.of(result);
+ } catch (Throwable primaryFailure) {
+ writer.abort(primaryFailure);
+ ExceptionUtils.rethrowException(primaryFailure);
+ return Optional.empty();
}
-
- List<ManifestFileMeta> merged = writer.result();
- result.addAll(merged);
- newFilesForAbort.addAll(merged);
- return Optional.of(result);
}
private static FullCompactionReadResult readForFullCompaction(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
index a083967e6b..e3f8c7af76 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
@@ -24,6 +24,7 @@ import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ExceptionUtils;
import javax.annotation.Nullable;
@@ -74,12 +75,18 @@ public class ManifestFileMerger {
}
return ManifestFileLegacyMerger.merge(
input, newFilesForAbort, manifestFile, partitionType,
options);
- } catch (Throwable e) {
+ } catch (Throwable primaryFailure) {
// exception occurs, clean up and rethrow
for (ManifestFileMeta manifest : newFilesForAbort) {
- manifestFile.delete(manifest.fileName());
+ try {
+ manifestFile.delete(manifest.fileName());
+ } catch (Throwable cleanupFailure) {
+ primaryFailure =
+ ExceptionUtils.firstOrSuppressed(cleanupFailure,
primaryFailure);
+ }
}
- throw new RuntimeException(e);
+ ExceptionUtils.rethrow(primaryFailure);
+ return null;
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
new file mode 100644
index 0000000000..d54e29e91f
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
@@ -0,0 +1,778 @@
+/*
+ * 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.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.CollectedDeletes;
+import org.apache.paimon.manifest.FileEntry.ReusableIdentifier;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.manifest.ManifestAvroReader;
+import org.apache.paimon.manifest.ManifestAvroWriter;
+import org.apache.paimon.manifest.ManifestAvroWriter.EncodedEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestFileMetaTestBase;
+import org.apache.paimon.manifest.ProjectedManifestEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.stats.StatsTestUtils;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.TraceableFileIO;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.AdditionalAnswers;
+import org.mockito.ArgumentMatchers;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+/** Tests cleanup when manifest rewrites fail with {@link Error}. */
+class ManifestRewriteCleanupTest extends ManifestFileMetaTestBase {
+
+ private static final RowType PARTITION_TYPE = RowType.of(DataTypes.INT());
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private FailingFileIO fileIO;
+ private ManifestFile manifestFile;
+ private Path manifestPath;
+
+ @BeforeEach
+ void beforeEach() {
+ fileIO = new FailingFileIO();
+ manifestFile = createManifestFile(tempDir.toString(), fileIO);
+ manifestPath = new Path(tempDir.toString(), "manifest");
+ }
+
+ @Test
+ void testRunMergeAbortsWriterAndPreservesError() throws Exception {
+ ManifestFileMeta input =
+ makeManifest(
+ rowIdEntry(FileKind.ADD, "file-0", 0),
+ rowIdEntry(FileKind.ADD, "file-1", 1));
+ int manifestCount = manifestFileCount();
+ AssertionError primaryFailure = new AssertionError("cursor failure");
+ IOException closeFailure = new IOException("cursor close failure");
+ List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
+ CollectedDeletes deletes = new CollectedDeletes(true);
+
+ Throwable thrown;
+ try {
+ ManifestEntryRunMergePlan plan =
+ runMergePlan(input, deletes, false, 1, primaryFailure,
closeFailure);
+ thrown = catchThrowable(() -> plan.mergeToManifest(manifestFile,
newFilesForAbort));
+ } finally {
+ deletes.release();
+ }
+
+
assertThat(thrown).isSameAs(primaryFailure).hasSuppressedException(closeFailure);
+ assertThat(newFilesForAbort).isEmpty();
+ assertNoManifestLeak(manifestCount);
+ }
+
+ @Test
+ void testRunMergeAbortsBothWritersWhenCleanupFails() throws Exception {
+ ManifestFileMeta input =
+ makeManifest(
+ rowIdEntry(FileKind.ADD, "add", 0),
+ rowIdEntry(FileKind.DELETE, "delete-0", 1),
+ rowIdEntry(FileKind.DELETE, "delete-1", 2));
+ int manifestCount = manifestFileCount();
+ AssertionError primaryFailure = new AssertionError("cursor failure");
+ List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
+ CollectedDeletes deletes = new CollectedDeletes(true);
+ fileIO.failDeletes();
+
+ Throwable thrown;
+ try {
+ ManifestEntryRunMergePlan plan =
+ runMergePlan(input, deletes, true, 2, primaryFailure,
null);
+ thrown =
+ catchThrowable(() ->
plan.mergeMinorToManifest(manifestFile, newFilesForAbort));
+ } finally {
+ deletes.release();
+ }
+
+ assertThat(thrown).isSameAs(primaryFailure);
+ assertThat(thrown.getSuppressed())
+ .extracting(Throwable::getMessage)
+ .containsExactly("delete failure 1", "delete failure 2");
+ assertThat(fileIO.deleteAttempts()).isEqualTo(2);
+ assertThat(newFilesForAbort).isEmpty();
+ assertNoManifestLeak(manifestCount);
+ }
+
+ @Test
+ void testExternalSortAbortsWriterAndPreservesError() throws Exception {
+ List<ManifestFileMeta> input =
+ Collections.singletonList(
+ makeManifest(
+ rowIdEntry(FileKind.ADD, "file-0", 0),
+ rowIdEntry(FileKind.ADD, "file-1", 1)));
+ int manifestCount = manifestFileCount();
+ AssertionError primaryFailure = new AssertionError("sort key failure");
+ List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
+ CollectedDeletes deletes = new CollectedDeletes(true);
+
+ Throwable thrown;
+ try {
+ thrown =
+ catchThrowable(
+ () ->
+
ManifestEntryExternalSort.sortAndWriteFullEntries(
+ input,
+ failingSortKey(input, 1,
primaryFailure),
+ externalSortConfig(),
+ manifestFile,
+ newFilesForAbort,
+ deletes,
+ 1));
+ } finally {
+ deletes.release();
+ }
+
+ assertThat(thrown).isSameAs(primaryFailure);
+ assertThat(newFilesForAbort).isEmpty();
+ assertNoManifestLeak(manifestCount);
+ }
+
+ @Test
+ void testExternalSortAbortsBothWritersWhenCleanupFails() throws Exception {
+ List<ManifestFileMeta> input =
+ Collections.singletonList(
+ makeManifest(
+ rowIdEntry(FileKind.ADD, "add", 0),
+ rowIdEntry(FileKind.DELETE, "delete-0", 1),
+ rowIdEntry(FileKind.DELETE, "delete-1", 2)));
+ int manifestCount = manifestFileCount();
+ AssertionError primaryFailure = new AssertionError("sort key failure");
+ List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
+ fileIO.failDeletes();
+
+ Throwable thrown =
+ catchThrowable(
+ () ->
+
ManifestEntryExternalSort.sortAndWriteMinorEntries(
+ input,
+ failingSortKey(input, 2,
primaryFailure),
+ externalSortConfig(),
+ manifestFile,
+ newFilesForAbort,
+ 1));
+
+ assertThat(thrown).isSameAs(primaryFailure);
+ assertThat(thrown.getSuppressed())
+ .extracting(Throwable::getMessage)
+ .containsExactly("delete failure 1", "delete failure 2");
+ assertThat(fileIO.deleteAttempts()).isEqualTo(2);
+ assertThat(newFilesForAbort).isEmpty();
+ assertNoManifestLeak(manifestCount);
+ }
+
+ @Test
+ void testLegacyMergeAbortsActiveWriterOnError() throws Exception {
+ List<ManifestFileMeta> input =
+ Arrays.asList(
+ makeManifest(makeEntry(true, "file-0")),
+ makeManifest(makeEntry(true, "file-1")));
+ int manifestCount = manifestFileCount();
+ AssertionError primaryFailure = new AssertionError("legacy writer
failure");
+ ManifestFile spyManifestFile = spy(manifestFile);
+ ManifestAvroWriter activeWriter = spy(manifestFile.createAvroWriter());
+ doReturn(activeWriter).when(spyManifestFile).createAvroWriter();
+ doAnswer(
+ invocation -> {
+ Iterable<? extends ManifestEntry> entries =
invocation.getArgument(0);
+ activeWriter.write(entries.iterator().next());
+ throw primaryFailure;
+ })
+ .when(activeWriter)
+ .write(ArgumentMatchers.<Iterable<? extends
ManifestEntry>>any());
+
+ Throwable thrown =
+ catchThrowable(
+ () ->
+ ManifestFileLegacyMerger.tryFullCompaction(
+ input,
+ new ArrayList<>(),
+ spyManifestFile,
+ Long.MAX_VALUE,
+ 1,
+ PARTITION_TYPE,
+ 1));
+
+ assertThat(thrown).isSameAs(primaryFailure);
+ assertNoManifestLeak(manifestCount);
+ }
+
+ @Test
+ void testBlockMergePreservesErrorWhenWriterCleanupFails() throws Exception
{
+ List<ManifestFileMeta> input =
+ Arrays.asList(
+ makeManifest(makeEntry(true, "file-0")),
+ makeManifest(makeEntry(true, "file-1")));
+ int manifestCount = manifestFileCount();
+ AssertionError primaryFailure = new AssertionError("block writer
failure");
+ ManifestFile spyManifestFile = spy(manifestFile);
+ ManifestAvroWriter activeWriter = spy(manifestFile.createAvroWriter());
+ doReturn(activeWriter).when(spyManifestFile).createAvroWriter();
+ doAnswer(
+ invocation -> {
+ activeWriter.write(makeEntry(true, "partial"));
+ throw primaryFailure;
+ })
+ .when(activeWriter)
+ .writeEncodedManifest(
+ ArgumentMatchers.any(ManifestAvroReader.class),
+ ArgumentMatchers.any(ManifestFileMeta.class));
+ fileIO.failDeletes();
+
+ Throwable thrown =
+ catchThrowable(
+ () ->
+ ManifestFileBlockMerger.tryFullCompaction(
+ input,
+ new ArrayList<>(),
+ spyManifestFile,
+ Long.MAX_VALUE,
+ 1,
+ PARTITION_TYPE,
+ 1));
+
+ assertThat(thrown).isSameAs(primaryFailure);
+ assertThat(thrown.getSuppressed())
+ .extracting(Throwable::getMessage)
+ .containsExactly("delete failure 1");
+ assertThat(fileIO.deleteAttempts()).isEqualTo(1);
+ assertNoManifestLeak(manifestCount);
+ }
+
+ @Test
+ void testWriterPreservesWriteErrorAndCleansAllRollingFiles() throws
Exception {
+ ManifestAvroWriter writer = createManifestFile(1).createAvroWriter();
+ ManifestEntry entry = rowIdEntry(FileKind.ADD, "file", 0);
+ // ManifestAvroWriter checks the rolling threshold every 1,000 records.
+ writer.write(Collections.nCopies(2000, entry));
+ assertThat(manifestFileCount()).isEqualTo(2);
+
+ AssertionError primaryFailure = new AssertionError("manifest write
failure");
+ AtomicInteger kindCalls = new AtomicInteger();
+ ManifestEntry failingEntry =
+ mock(ManifestEntry.class,
AdditionalAnswers.delegatesTo(entry));
+ doAnswer(
+ invocation -> {
+ if (kindCalls.incrementAndGet() == 2) {
+ throw primaryFailure;
+ }
+ return entry.kind();
+ })
+ .when(failingEntry)
+ .kind();
+ fileIO.failNextDeleteBeforeDeletion();
+
+ Throwable thrown = catchThrowable(() -> writer.write(failingEntry));
+
+ assertThat(thrown).isSameAs(primaryFailure);
+ assertThat(thrown.getSuppressed())
+ .extracting(Throwable::getMessage)
+ .containsExactly("delete failure 1");
+ assertThat(fileIO.deleteAttempts()).isEqualTo(3);
+ assertThat(manifestFileCount()).isEqualTo(1);
+ assertThat(
+ TraceableFileIO.openOutputStreams(
+ path ->
path.toString().startsWith(manifestPath.toString())))
+ .isEmpty();
+ }
+
+ @Test
+ void testCloseCursorsContinuesAfterError() {
+ AssertionError primaryFailure = new AssertionError("first close
failure");
+ IOException cleanupFailure = new IOException("second close failure");
+ CloseFailureCursor first = new CloseFailureCursor(primaryFailure);
+ CloseFailureCursor second = new CloseFailureCursor(cleanupFailure);
+ CloseFailureCursor third = new CloseFailureCursor(null);
+
+ Throwable thrown =
+ catchThrowable(
+ () ->
+ ManifestEntryRunMergePlan.closeCursors(
+ Arrays.asList(first, second, third)));
+
+
assertThat(thrown).isSameAs(primaryFailure).hasSuppressedException(cleanupFailure);
+ assertThat(first.closed).isTrue();
+ assertThat(second.closed).isTrue();
+ assertThat(third.closed).isTrue();
+ }
+
+ @Test
+ void testManifestFileMergerPreservesErrorWhenCleanupFails() throws
IOException {
+ List<ManifestFileMeta> input =
+ Arrays.asList(
+ makeManifest(makeEntry(true, "file-0")),
+ makeManifest(makeEntry(true, "file-1")),
+ makeManifest(makeEntry(true, "file-2")),
+ makeManifest(makeEntry(true, "file-3")));
+ int manifestCount = manifestFileCount();
+ long targetSize =
+
input.stream().mapToLong(ManifestFileMeta::fileSize).max().getAsLong() + 1;
+ AssertionError primaryFailure = new AssertionError("manifest merge
failure");
+ fileIO.failDeletes();
+
+ Options options = new Options();
+ options.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), targetSize +
"B");
+ options.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
Long.MAX_VALUE + "B");
+ options.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 2);
+ options.set(CoreOptions.MANIFEST_MERGE_OPTIMIZE_ENABLED, false);
+ options.set(CoreOptions.SCAN_MANIFEST_PARALLELISM, 1);
+
+ // The first iteration checks full compaction. Fail in the second
iteration after
+ // compactMinor registers one file.
+ List<ManifestFileMeta> failingInput = new
FailingSecondIterationList(input, primaryFailure);
+ Throwable thrown =
+ catchThrowable(
+ () ->
+ ManifestFileMerger.merge(
+ failingInput,
+ manifestFile,
+ PARTITION_TYPE,
+ new CoreOptions(options)));
+
+ assertThat(thrown).isSameAs(primaryFailure);
+ assertThat(thrown.getSuppressed())
+ .extracting(Throwable::getMessage)
+ .containsExactly("delete failure 1");
+ assertThat(fileIO.deleteAttempts()).isEqualTo(1);
+ assertNoManifestLeak(manifestCount);
+ }
+
+ private ManifestEntryRunMergePlan runMergePlan(
+ ManifestFileMeta input,
+ CollectedDeletes deletes,
+ boolean minor,
+ int successfulAdvancesBeforeFailure,
+ AssertionError failure,
+ IOException closeFailure) {
+ ManifestEntryRunMerge.SortPartitionDictionary partitions =
+ new ManifestEntryRunMerge.SortPartitionDictionary(
+ (left, right) -> Integer.compare(left.getInt(0),
right.getInt(0)));
+ ManifestEntryRunMergePlan.Source.Spec source =
+ (file, planDeletes, planMinor, planPartitions) ->
+ new FailingCursor(
+ new
ManifestEntryRunMergePlan.InMemoryManifestCursor(
+ file, input, planDeletes, planMinor,
planPartitions),
+ successfulAdvancesBeforeFailure,
+ failure,
+ closeFailure);
+ return new ManifestEntryRunMergePlan(
+ Collections.singletonList(source), partitions, deletes, minor);
+ }
+
+ private ManifestFileSorter.ManifestSortKey failingSortKey(
+ List<ManifestFileMeta> input, int successfulRowsBeforeFailure,
AssertionError failure) {
+ return new FailingManifestSortKey(
+ ManifestFileSorter.createSortKey(true, input, null,
PARTITION_TYPE),
+ successfulRowsBeforeFailure,
+ failure);
+ }
+
+ private ManifestEntryExternalSort.ExternalSortConfig externalSortConfig() {
+ Options options = new Options();
+ options.set(CoreOptions.SORT_SPILL_BUFFER_SIZE.key(), "1 mb");
+ return ManifestEntryExternalSort.ExternalSortConfig.from(new
CoreOptions(options), null);
+ }
+
+ private ManifestEntry rowIdEntry(FileKind kind, String fileName, long
firstRowId) {
+ BinaryRow partition = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(partition);
+ writer.writeInt(0, 0);
+ writer.complete();
+ return ManifestEntry.create(
+ kind,
+ partition,
+ 0,
+ 0,
+ DataFileMeta.create(
+ fileName,
+ 0,
+ 1,
+ partition,
+ partition,
+ StatsTestUtils.newEmptySimpleStats(),
+ StatsTestUtils.newEmptySimpleStats(),
+ 0,
+ 0,
+ 0,
+ 0,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(200000),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ Collections.singletonList("f0")));
+ }
+
+ private ManifestFile createManifestFile(long suggestedFileSize) {
+ Path tablePath = new Path(tempDir.toString());
+ return new ManifestFile.Factory(
+ fileIO,
+ new SchemaManager(fileIO, tablePath),
+ PARTITION_TYPE,
+ avro,
+ "zstd",
+ new FileStorePathFactory(
+ tablePath,
+ PARTITION_TYPE,
+ "default",
+ CoreOptions.FILE_FORMAT.defaultValue(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+
CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null,
+ null,
+ CoreOptions.ExternalPathStrategy.NONE,
+ null,
+ false,
+ null),
+ suggestedFileSize,
+ null)
+ .create();
+ }
+
+ private int manifestFileCount() throws IOException {
+ return fileIO.listStatus(manifestPath).length;
+ }
+
+ private void assertNoManifestLeak(int expectedManifestCount) throws
IOException {
+ assertThat(manifestFileCount()).isEqualTo(expectedManifestCount);
+ assertThat(
+ TraceableFileIO.openOutputStreams(
+ path ->
path.toString().startsWith(manifestPath.toString())))
+ .isEmpty();
+ }
+
+ @Override
+ protected ManifestFile getManifestFile() {
+ return manifestFile;
+ }
+
+ @Override
+ protected RowType getPartitionType() {
+ return PARTITION_TYPE;
+ }
+
+ private static class FailingCursor implements
ManifestEntryRunMergePlan.Cursor {
+
+ private final ManifestEntryRunMergePlan.Cursor delegate;
+ private final int successfulAdvancesBeforeFailure;
+ private final AssertionError failure;
+ private final IOException closeFailure;
+ private int successfulAdvances;
+
+ private FailingCursor(
+ ManifestEntryRunMergePlan.Cursor delegate,
+ int successfulAdvancesBeforeFailure,
+ AssertionError failure,
+ IOException closeFailure) {
+ this.delegate = delegate;
+ this.successfulAdvancesBeforeFailure =
successfulAdvancesBeforeFailure;
+ this.failure = failure;
+ this.closeFailure = closeFailure;
+ }
+
+ @Override
+ public boolean advance() throws Exception {
+ if (successfulAdvances >= successfulAdvancesBeforeFailure) {
+ throw failure;
+ }
+ boolean advanced = delegate.advance();
+ if (advanced) {
+ successfulAdvances++;
+ }
+ return advanced;
+ }
+
+ @Override
+ public boolean hasCurrent() {
+ return delegate.hasCurrent();
+ }
+
+ @Override
+ public ProjectedManifestEntry current() {
+ return delegate.current();
+ }
+
+ @Override
+ public EncodedEntry metadata() {
+ return delegate.metadata();
+ }
+
+ @Override
+ public ManifestEntryRunMerge.SortKey key() {
+ return delegate.key();
+ }
+
+ @Override
+ public ByteBuffer encodedRecord() {
+ return delegate.encodedRecord();
+ }
+
+ @Override
+ public ReusableIdentifier identifier() {
+ return delegate.identifier();
+ }
+
+ @Override
+ public void close() throws Exception {
+ delegate.close();
+ if (closeFailure != null) {
+ throw closeFailure;
+ }
+ }
+ }
+
+ private static class FailingManifestSortKey implements
ManifestFileSorter.ManifestSortKey {
+
+ private final ManifestFileSorter.ManifestSortKey delegate;
+ private final int successfulRowsBeforeFailure;
+ private final AssertionError failure;
+ private int successfulRows;
+
+ private FailingManifestSortKey(
+ ManifestFileSorter.ManifestSortKey delegate,
+ int successfulRowsBeforeFailure,
+ AssertionError failure) {
+ this.delegate = delegate;
+ this.successfulRowsBeforeFailure = successfulRowsBeforeFailure;
+ this.failure = failure;
+ }
+
+ @Override
+ public int compareMin(ManifestFileMeta left, ManifestFileMeta right) {
+ return delegate.compareMin(left, right);
+ }
+
+ @Override
+ public int compareMax(ManifestFileMeta left, ManifestFileMeta right) {
+ return delegate.compareMax(left, right);
+ }
+
+ @Override
+ public boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta
maxFile) {
+ return delegate.isAfterMax(file, maxFile);
+ }
+
+ @Override
+ public RowType externalSortRowType() {
+ return delegate.externalSortRowType();
+ }
+
+ @Override
+ public int[] externalSortKeyFields() {
+ return delegate.externalSortKeyFields();
+ }
+
+ @Override
+ public void replaceExternalSortRow(
+ GenericRow row, ManifestEntry entry, InternalRow
binaryManifestRow) {
+ delegate.replaceExternalSortRow(row, entry, binaryManifestRow);
+ }
+
+ @Override
+ public InternalRow binaryManifestRow(BinaryRow row) {
+ if (successfulRows >= successfulRowsBeforeFailure) {
+ throw failure;
+ }
+ successfulRows++;
+ return delegate.binaryManifestRow(row);
+ }
+ }
+
+ private static class CloseFailureCursor implements
ManifestEntryRunMergePlan.Cursor {
+
+ private final Throwable failure;
+ private boolean closed;
+
+ private CloseFailureCursor(Throwable failure) {
+ this.failure = failure;
+ }
+
+ @Override
+ public boolean advance() {
+ return false;
+ }
+
+ @Override
+ public boolean hasCurrent() {
+ return false;
+ }
+
+ @Override
+ public ProjectedManifestEntry current() {
+ return null;
+ }
+
+ @Override
+ public EncodedEntry metadata() {
+ return null;
+ }
+
+ @Override
+ public ManifestEntryRunMerge.SortKey key() {
+ return null;
+ }
+
+ @Override
+ public ByteBuffer encodedRecord() {
+ return null;
+ }
+
+ @Override
+ public ReusableIdentifier identifier() {
+ return null;
+ }
+
+ @Override
+ public void close() throws Exception {
+ closed = true;
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure != null) {
+ throw (Exception) failure;
+ }
+ }
+ }
+
+ private static class FailingFileIO extends TraceableFileIO {
+
+ private boolean trackDeletes;
+ private boolean failDeletesAfterDeletion;
+ private int failuresBeforeDeletion;
+ private int deleteAttempts;
+
+ private void failDeletes() {
+ trackDeletes = true;
+ failDeletesAfterDeletion = true;
+ }
+
+ private void failNextDeleteBeforeDeletion() {
+ trackDeletes = true;
+ failuresBeforeDeletion = 1;
+ }
+
+ private int deleteAttempts() {
+ return deleteAttempts;
+ }
+
+ @Override
+ public boolean delete(Path file, boolean recursive) throws IOException
{
+ if (trackDeletes) {
+ deleteAttempts++;
+ if (failuresBeforeDeletion > 0) {
+ failuresBeforeDeletion--;
+ throw new RuntimeException("delete failure " +
deleteAttempts);
+ }
+ }
+ boolean deleted = super.delete(file, recursive);
+ if (failDeletesAfterDeletion) {
+ throw new RuntimeException("delete failure " + deleteAttempts);
+ }
+ return deleted;
+ }
+ }
+
+ private static class FailingSecondIterationList extends
AbstractList<ManifestFileMeta> {
+
+ private final List<ManifestFileMeta> delegate;
+ private final AssertionError failure;
+ private int iterationCount;
+
+ private FailingSecondIterationList(
+ List<ManifestFileMeta> delegate, AssertionError failure) {
+ this.delegate = delegate;
+ this.failure = failure;
+ }
+
+ @Override
+ public ManifestFileMeta get(int index) {
+ return delegate.get(index);
+ }
+
+ @Override
+ public int size() {
+ return delegate.size();
+ }
+
+ @Override
+ public Iterator<ManifestFileMeta> iterator() {
+ Iterator<ManifestFileMeta> iterator = delegate.iterator();
+ int currentIteration = ++iterationCount;
+ return new Iterator<ManifestFileMeta>() {
+
+ private int returned;
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public ManifestFileMeta next() {
+ if (currentIteration == 2 && returned == 2) {
+ throw failure;
+ }
+ returned++;
+ return iterator.next();
+ }
+ };
+ }
+ }
+}
