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 2e8cf7779e [common] Release the BTree index file on every path that
abandons the writer (#9474)
2e8cf7779e is described below
commit 2e8cf7779e460a34ad0d3920f99a0716d18f18dc
Author: ZIHAN DAI <[email protected]>
AuthorDate: Sun Sep 20 15:10:51 2026 +1000
[common] Release the BTree index file on every path that abandons the
writer (#9474)
---
.../paimon/globalindex/btree/BTreeIndexWriter.java | 34 +++-
.../btree/BTreeIndexWriterCloseTest.java | 204 +++++++++++++++++++++
2 files changed, 233 insertions(+), 5 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
index b8d6f62bea..fecf600f56 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
@@ -31,12 +31,14 @@ import org.apache.paimon.sst.BlockHandle;
import org.apache.paimon.sst.BloomFilterHandle;
import org.apache.paimon.sst.SstFileWriter;
import org.apache.paimon.utils.BloomFilter;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.LazyField;
import org.apache.paimon.utils.LongArrayList;
import org.apache.paimon.utils.RoaringNavigableMap64;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.IOException;
import java.util.Collections;
import java.util.Comparator;
@@ -70,10 +72,11 @@ import java.util.zip.CRC32;
* <p>For efficiency, we combine entries with the same keys and store a
compact list of row ids for
* each key.
*/
-public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter {
+public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter,
Closeable {
private final String fileName;
private final PositionOutputStream out;
+ private boolean closed;
private final SstFileWriter writer;
private final KeySerializer keySerializer;
@@ -126,10 +129,15 @@ public class BTreeIndexWriter implements
GlobalIndexSingleColumnWriter {
}
this.fileName =
indexFileWriter.newFileName(BTreeGlobalIndexerFactory.IDENTIFIER);
this.out = indexFileWriter.newOutputStream(this.fileName);
- this.keySerializer = keySerializer;
- this.comparator = keySerializer.createComparator();
- this.fileVersion = fileVersion;
- this.writer = new SstFileWriter(out, blockSize, bloomFilterBuilder,
compressionFactory);
+ try {
+ this.keySerializer = keySerializer;
+ this.comparator = keySerializer.createComparator();
+ this.fileVersion = fileVersion;
+ this.writer = new SstFileWriter(out, blockSize,
bloomFilterBuilder, compressionFactory);
+ } catch (RuntimeException | Error e) {
+ IOUtils.closeQuietly(out);
+ throw e;
+ }
}
@Override
@@ -205,7 +213,10 @@ public class BTreeIndexWriter implements
GlobalIndexSingleColumnWriter {
writer.writeSlice(footerEncoding);
out.close();
+ closed = true;
} catch (IOException e) {
+ IOUtils.closeQuietly(out);
+ closed = true;
throw new RuntimeException("Error in closing BTree index writer",
e);
}
@@ -243,4 +254,17 @@ public class BTreeIndexWriter implements
GlobalIndexSingleColumnWriter {
return nullBitmapHandle;
}
+
+ /**
+ * Releases the output stream for a build that is abandoned without {@link
#finish()}. The owner
+ * cleanup paths reach a writer only through {@code instanceof
AutoCloseable}, so without this
+ * the stream opened in the constructor stays open for the life of the
process.
+ */
+ @Override
+ public void close() throws IOException {
+ if (!closed) {
+ closed = true;
+ out.close();
+ }
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexWriterCloseTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexWriterCloseTest.java
new file mode 100644
index 0000000000..b351db26b8
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexWriterCloseTest.java
@@ -0,0 +1,204 @@
+/*
+ * 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.globalindex.btree;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.utils.IOUtils;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests that {@link BTreeIndexWriter} releases the file it opened, on every
path that abandons it.
+ */
+public class BTreeIndexWriterCloseTest {
+
+ /**
+ * finish() writes the null bitmap, the bloom filter, the index block and
the footer before
+ * closing. A write that fails part way through - a full disk is the
ordinary cause - must not
+ * leave the file open.
+ */
+ @Test
+ public void testFinishReleasesTheFileWhenWritingFails() throws IOException
{
+ AtomicInteger closed = new AtomicInteger();
+ BTreeIndexWriter writer =
+ new BTreeIndexWriter(
+ failingWriter(closed),
+ KeySerializer.create(new IntType()),
+ 1024,
+ (BlockCompressionFactory) null);
+
+ for (int i = 0; i < 200; i++) {
+ writer.write(i, (long) i);
+ }
+
+ assertThatThrownBy(writer::finish)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Error in closing BTree index writer");
+
+ assertThat(closed).hasValue(1);
+ }
+
+ /**
+ * A build can be abandoned before finish() ever runs - the row source
fails, or a Flink or
+ * Spark index build task is cancelled. The owner cleanup paths,
PkSortedIndexFile#build and
+ * SortedSingleColumnIndexWriter#close, release the writer only if they
can see it as an
+ * AutoCloseable, and skip it silently otherwise. This is that exact idiom.
+ */
+ @Test
+ public void testTheOwnerCleanupPathReleasesAnAbandonedWriter() throws
IOException {
+ AtomicInteger closed = new AtomicInteger();
+ BTreeIndexWriter writer =
+ new BTreeIndexWriter(
+ failingWriter(closed),
+ KeySerializer.create(new IntType()),
+ 1024,
+ (BlockCompressionFactory) null);
+ writer.write(1, 1L);
+
+ if (writer instanceof AutoCloseable) {
+ IOUtils.closeQuietly((AutoCloseable) writer);
+ }
+
+ assertThat(closed).hasValue(1);
+ }
+
+ /**
+ * SortedGlobalIndexWriter holds the task writer in a try-with-resources,
so close() can follow
+ * a successful finish(). It must not close the stream a second time.
+ */
+ @Test
+ public void testCloseIsIdempotent() throws IOException {
+ AtomicInteger closed = new AtomicInteger();
+ BTreeIndexWriter writer =
+ new BTreeIndexWriter(
+ failingWriter(closed),
+ KeySerializer.create(new IntType()),
+ 1024,
+ (BlockCompressionFactory) null);
+ writer.write(1, 1L);
+
+ writer.close();
+ writer.close();
+
+ assertThat(closed).hasValue(1);
+ }
+
+ /**
+ * SstFileWriter allocates its block buffers after the stream is open, so
a direct or heap
+ * allocation failure there arrives as an Error rather than an exception.
The stream still has
+ * to be released: a task runtime can catch that and keep the JVM alive.
+ */
+ @Test
+ public void testConstructorReleasesTheFileWhenSetupThrowsAnError() {
+ AtomicInteger closed = new AtomicInteger();
+ assertThatThrownBy(
+ () ->
+ new BTreeIndexWriter(
+ failingWriter(closed),
+ errorThrowingSerializer(),
+ 1024,
+ (BlockCompressionFactory) null))
+ .isInstanceOf(OutOfMemoryError.class);
+
+ assertThat(closed).hasValue(1);
+ }
+
+ private static KeySerializer errorThrowingSerializer() {
+ KeySerializer delegate = KeySerializer.create(new IntType());
+ return new KeySerializer() {
+
+ @Override
+ public byte[] serialize(Object key) {
+ return delegate.serialize(key);
+ }
+
+ @Override
+ public Object deserialize(MemorySlice data) {
+ return delegate.deserialize(data);
+ }
+
+ @Override
+ public Comparator<Object> createComparator() {
+ throw new OutOfMemoryError("Direct buffer memory");
+ }
+ };
+ }
+
+ private static GlobalIndexFileWriter failingWriter(AtomicInteger closed) {
+ return new GlobalIndexFileWriter() {
+
+ @Override
+ public String newFileName(String prefix) {
+ return "test-btree" + prefix;
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) {
+ return new PositionOutputStream() {
+
+ private long pos;
+ private final long capacity = 1500L;
+
+ @Override
+ public long getPos() {
+ return pos;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ write(new byte[] {(byte) b}, 0, 1);
+ }
+
+ @Override
+ public void write(byte[] b) throws IOException {
+ write(b, 0, b.length);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws
IOException {
+ if (pos + len > capacity) {
+ throw new IOException("no space left on device");
+ }
+ pos += len;
+ }
+
+ @Override
+ public void flush() {}
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ }
+ };
+ }
+ };
+ }
+}
