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 045ffd9d6e [core] Fix OOM error handling in MergeTreeWriter and
improve resource cleanup (#9933)
045ffd9d6e is described below
commit 045ffd9d6e1e79e04dcfc36ab5307e25a62d2738
Author: Alex <[email protected]>
AuthorDate: Fri Sep 18 19:34:14 2026 +0800
[core] Fix OOM error handling in MergeTreeWriter and improve resource
cleanup (#9933)
---
.../apache/paimon/io/RollingFileWriterImpl.java | 11 +++--
.../apache/paimon/io/RollingFileWriterTest.java | 57 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 3 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
index 11c332b90f..1d8e0fc158 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
@@ -49,6 +49,7 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
private long recordCount = 0;
private long currentFileRecordCount = 0;
private boolean closed = false;
+ private boolean aborted = false;
public RollingFileWriterImpl(
Supplier<? extends SingleFileWriter<T, R>> writerFactory,
@@ -177,8 +178,12 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
@Override
public void abort() {
- if (currentWriter != null) {
- currentWriter.abort();
+ SingleFileWriter<T, R> writer = currentWriter;
+ currentWriter = null;
+ aborted = true;
+
+ if (writer != null) {
+ writer.abort();
}
for (FileWriterAbortExecutor abortExecutor : closedWriters) {
abortExecutor.abort();
@@ -201,7 +206,7 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
@Override
public void close() throws IOException {
- if (closed) {
+ if (closed || aborted) {
return;
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
index 1676d7ebd4..175fc09a7e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
@@ -24,6 +24,8 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.manifest.FileSource;
@@ -48,8 +50,11 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link RollingFileWriterImpl}. */
public class RollingFileWriterTest {
@@ -140,6 +145,44 @@ public class RollingFileWriterTest {
}
}
+ @Test
+ public void testCloseAfterWriteFailure() throws IOException {
+ Path path = new Path(tempDir.resolve("failed-file").toString());
+ OutOfMemoryError expected = new OutOfMemoryError("expected");
+ RollingFileWriterImpl<InternalRow, Void> writer =
+ new RollingFileWriterImpl<>(
+ () ->
+ new TestSingleFileWriter(
+ LocalFileIO.create(),
+ (out, compression) ->
+ new FormatWriter() {
+ @Override
+ public void
addElement(InternalRow element) {
+ throw expected;
+ }
+
+ @Override
+ public boolean
reachTargetSize(
+ boolean
suggestedCheck,
+ long targetSize) {
+ return false;
+ }
+
+ @Override
+ public void close() {}
+ },
+ path),
+ Long.MAX_VALUE,
+ Long.MAX_VALUE);
+
+ assertThatThrownBy(() ->
writer.write(GenericRow.of(1))).isSameAs(expected);
+ assertThat(LocalFileIO.create().exists(path)).isFalse();
+ assertThatCode(writer::abort).doesNotThrowAnyException();
+ assertThatCode(writer::close).doesNotThrowAnyException();
+
assertThatThrownBy(writer::result).isInstanceOf(IllegalStateException.class);
+
assertThatThrownBy(writer::drainAbortExecutors).isInstanceOf(IllegalStateException.class);
+ }
+
@Test
public void testRollingByRows() throws IOException {
// Huge byte target so only the row-count limit triggers rolling.
@@ -313,6 +356,20 @@ public class RollingFileWriterTest {
assertThat(file.valueStats().minValues().getFieldCount()).isEqualTo(SCHEMA.getFieldCount());
}
+ private static class TestSingleFileWriter extends
SingleFileWriter<InternalRow, Void> {
+
+ private TestSingleFileWriter(
+ LocalFileIO fileIO, FormatWriterFactory writerFactory, Path
path) {
+ super(fileIO, writerFactory, path, Function.identity(), null,
false);
+ }
+
+ @Override
+ public Void result() throws IOException {
+ fileIO.getFileSize(path);
+ return null;
+ }
+ }
+
private static class SingleUseBundleRecords implements BundleRecords {
private final List<InternalRow> rows;
