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 93bcff41af [full-text] Fix native reader cleanup on failures (#8950)
93bcff41af is described below
commit 93bcff41af389ad629941120537582abad3fec8a
Author: QuakeWang <[email protected]>
AuthorDate: Fri Jul 31 10:12:54 2026 +0800
[full-text] Fix native reader cleanup on failures (#8950)
---
paimon-full-text/pom.xml | 14 +++
.../index/NativeFullTextGlobalIndexReader.java | 35 ++++--
.../index/NativeFullTextGlobalIndexReaderTest.java | 133 +++++++++++++++++++++
3 files changed, 175 insertions(+), 7 deletions(-)
diff --git a/paimon-full-text/pom.xml b/paimon-full-text/pom.xml
index 7dcf4ad9a2..b2a40710ee 100644
--- a/paimon-full-text/pom.xml
+++ b/paimon-full-text/pom.xml
@@ -53,6 +53,20 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-inline</artifactId>
+ <version>${mockito.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${mockito.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.paimon</groupId>
<artifactId>paimon-core</artifactId>
diff --git
a/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexReader.java
b/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexReader.java
index 8c292dd898..4df87f6c97 100644
---
a/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexReader.java
+++
b/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexReader.java
@@ -29,6 +29,7 @@ import org.apache.paimon.index.fulltext.FullTextIndexReader;
import org.apache.paimon.index.fulltext.FullTextSearchResult;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.FullTextSearch;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.RoaringNavigableMap64;
import java.io.IOException;
@@ -112,9 +113,9 @@ public class NativeFullTextGlobalIndexReader implements
GlobalIndexReader {
try {
inputStream = input;
reader = new FullTextIndexReader(new
PaimonFullTextIndexInput(input));
- } catch (RuntimeException e) {
- input.close();
+ } catch (RuntimeException | Error e) {
inputStream = null;
+ closeResource(input, e);
throw e;
}
}
@@ -136,14 +137,34 @@ public class NativeFullTextGlobalIndexReader implements
GlobalIndexReader {
@Override
public void close() throws IOException {
- if (reader != null) {
- reader.close();
+ Throwable failure = null;
+ try {
+ failure = closeResource(reader, failure);
+ failure = closeResource(inputStream, failure);
+ } finally {
reader = null;
- }
- if (inputStream != null) {
- inputStream.close();
inputStream = null;
}
+ if (failure instanceof IOException) {
+ throw (IOException) failure;
+ } else if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ } else if (failure instanceof Error) {
+ throw (Error) failure;
+ } else if (failure != null) {
+ throw new IOException("Failed to close native full-text index
reader", failure);
+ }
+ }
+
+ private static Throwable closeResource(AutoCloseable resource, Throwable
failure) {
+ try {
+ if (resource != null) {
+ resource.close();
+ }
+ } catch (Throwable closeFailure) {
+ return ExceptionUtils.firstOrSuppressed(closeFailure, failure);
+ }
+ return failure;
}
private static class PaimonFullTextIndexInput implements
FullTextIndexInput {
diff --git
a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexReaderTest.java
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexReaderTest.java
new file mode 100644
index 0000000000..13547fe9aa
--- /dev/null
+++
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexReaderTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.fulltext.index;
+
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.index.fulltext.FullTextIndexReader;
+import org.apache.paimon.predicate.FullTextSearch;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.concurrent.CompletionException;
+
+import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.catchThrowable;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for resource lifecycle of {@link NativeFullTextGlobalIndexReader}. */
+public class NativeFullTextGlobalIndexReaderTest {
+
+ @Test
+ public void
testInitializationFailurePreservesOriginalExceptionWhenCleanupThrowsUnchecked()
{
+ RuntimeException closeFailure = new RuntimeException("input close
failure");
+ TestingSeekableInputStream input =
+ new TestingSeekableInputStream(new byte[64], closeFailure);
+ NativeFullTextGlobalIndexReader reader = createReader(input);
+
+ Throwable failure =
+ catchThrowable(
+ () ->
+ reader.visitFullTextSearch(new
FullTextSearch("text", "{}", 1))
+ .join());
+
+ assertThat(failure).isInstanceOf(CompletionException.class);
+ Throwable initializationFailure = failure.getCause();
+ assertThat(initializationFailure)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("invalid storage format");
+
assertThat(initializationFailure.getSuppressed()).containsExactly(closeFailure);
+ assertThat(input.closeCalls).isEqualTo(1);
+
+ assertThatCode(reader::close).doesNotThrowAnyException();
+ assertThat(input.closeCalls).isEqualTo(1);
+ }
+
+ @Test
+ public void testCloseAttemptsAllResourcesWhenReaderCloseThrowsError()
throws Exception {
+ NativeFullTextGlobalIndexReader reader =
+ createReader(new TestingSeekableInputStream(new byte[1],
null));
+ FullTextIndexReader nativeReader = mock(FullTextIndexReader.class);
+ UnsatisfiedLinkError readerCloseFailure = new
UnsatisfiedLinkError("reader close failure");
+ doThrow(readerCloseFailure).when(nativeReader).close();
+ IOException inputCloseFailure = new IOException("input close failure");
+ TestingSeekableInputStream input =
+ new TestingSeekableInputStream(new byte[1], inputCloseFailure);
+ setResource(reader, "reader", nativeReader);
+ setResource(reader, "inputStream", input);
+
+ Throwable failure = catchThrowable(reader::close);
+
+ assertThat(failure).isSameAs(readerCloseFailure);
+ assertThat(failure.getSuppressed()).containsExactly(inputCloseFailure);
+ assertThat(input.closeCalls).isEqualTo(1);
+
+ assertThatCode(reader::close).doesNotThrowAnyException();
+ verify(nativeReader, times(1)).close();
+ assertThat(input.closeCalls).isEqualTo(1);
+ }
+
+ private NativeFullTextGlobalIndexReader
createReader(TestingSeekableInputStream input) {
+ GlobalIndexIOMeta meta =
+ new GlobalIndexIOMeta(new Path("file:///unused"), 64L, new
byte[0]);
+ return new NativeFullTextGlobalIndexReader(
+ ignored -> input, Collections.singletonList(meta),
newDirectExecutorService());
+ }
+
+ private static void setResource(
+ NativeFullTextGlobalIndexReader reader, String fieldName, Object
resource)
+ throws ReflectiveOperationException {
+ Field field =
NativeFullTextGlobalIndexReader.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(reader, resource);
+ }
+
+ private static class TestingSeekableInputStream extends
ByteArraySeekableStream {
+
+ private final Throwable closeFailure;
+ private int closeCalls;
+
+ private TestingSeekableInputStream(byte[] bytes, Throwable
closeFailure) {
+ super(bytes);
+ this.closeFailure = closeFailure;
+ }
+
+ @Override
+ public void close() throws IOException {
+ closeCalls++;
+ if (closeFailure instanceof IOException) {
+ throw (IOException) closeFailure;
+ } else if (closeFailure instanceof RuntimeException) {
+ throw (RuntimeException) closeFailure;
+ } else if (closeFailure instanceof Error) {
+ throw (Error) closeFailure;
+ }
+ super.close();
+ }
+ }
+}
