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 97e9e3b3e8 [core] Keep shared IOManager open after clustering sort
(#10068)
97e9e3b3e8 is described below
commit 97e9e3b3e8fd06e09627e4aac59429062cd739e0
Author: hutiefang76 <[email protected]>
AuthorDate: Tue Sep 22 10:43:24 2026 +0800
[core] Keep shared IOManager open after clustering sort (#10068)
---
.../org/apache/paimon/append/cluster/Sorter.java | 9 ++--
.../apache/paimon/append/cluster/SorterTest.java | 55 ++++++++++++++++++++++
2 files changed, 58 insertions(+), 6 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
index e1cfe9fbe3..bcfcf4fc61 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
@@ -46,7 +46,6 @@ public abstract class Sorter {
protected final int[] valueProjectionMap;
private final int arity;
- private final transient IOManager ioManager;
private final transient BinaryExternalSortBuffer buffer;
public Sorter(
@@ -77,7 +76,6 @@ public abstract class Sorter {
CompressOptions spillCompression = options.spillCompressOptions();
MemorySize maxDiskSize = options.writeBufferSpillDiskSize();
- this.ioManager = ioManager;
this.buffer =
BinaryExternalSortBuffer.create(
ioManager,
@@ -116,11 +114,10 @@ public abstract class Sorter {
}
public void close() throws Exception {
- if (buffer != null) {
+ try {
buffer.clear();
- }
- if (ioManager != null) {
- ioManager.close();
+ } finally {
+ reader.close();
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/SorterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/SorterTest.java
index 8b31531032..db9f38de10 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/cluster/SorterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/cluster/SorterTest.java
@@ -24,8 +24,10 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
@@ -37,6 +39,7 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.MutableObjectIterator;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -62,6 +65,28 @@ public class SorterTest {
innerTest(curve);
}
+ @Test
+ public void testCloseKeepsCallerOwnedIOManagerOpenAndClosesReader() throws
Exception {
+ FileStoreTable table = createTable(new HashMap<>(), "order");
+ TrackingIOManager ioManager = new
TrackingIOManager(ioTempDir.toString());
+ TrackingRecordReader reader = new TrackingRecordReader();
+ Sorter sorter =
+ Sorter.getSorter(
+ new RecordReaderIterator<>(reader),
+ ioManager,
+ table.rowType(),
+ table.coreOptions());
+
+ try {
+ sorter.close();
+
+ assertThat(ioManager.closed).isFalse();
+ assertThat(reader.closed).isTrue();
+ } finally {
+ ioManager.close();
+ }
+ }
+
private void innerTest(String curve) throws Exception {
FileStoreTable table = createTable(new HashMap<>(), curve);
writeOnce(
@@ -138,4 +163,34 @@ public class SorterTest {
new Path(tableTempDir.toString()),
schemaManager.createTable(schema));
}
+
+ private static class TrackingIOManager extends IOManagerImpl {
+
+ private boolean closed;
+
+ private TrackingIOManager(String tempDir) {
+ super(tempDir);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closed = true;
+ super.close();
+ }
+ }
+
+ private static class TrackingRecordReader implements
RecordReader<InternalRow> {
+
+ private boolean closed;
+
+ @Override
+ public RecordIterator<InternalRow> readBatch() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
}
