[core] Should not close channelManager in BinaryExternalSortBuffer (#5466)

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java

@@ -50,7 +50,7 @@ public class BinaryExternalSortBuffer implements SortBuffer {
     private final BinaryRowSerializer serializer;
     private final BinaryInMemorySortBuffer inMemorySortBuffer;
     private final IOManager ioManager;
-    private SpillChannelManager channelManager;
+    private final SpillChannelManager channelManager;
     private final int maxNumFileHandles;
     private final BlockCompressionFactory compressionCodecFactory;
     private final int compressionBlockSize;
@@ -154,8 +154,7 @@ public void clear() {
         inMemorySortBuffer.clear();
         spillChannelIDs.clear();
         // delete files
-        channelManager.close();
-        channelManager = new SpillChannelManager();
+        channelManager.reset();
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/sort/SpillChannelManager.java

@@ -20,56 +20,40 @@
 
 import org.apache.paimon.disk.FileIOChannel;
 
-import java.io.Closeable;
 import java.io.File;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
 /** Channel manager to manage the life cycle of spill channels. */
-public class SpillChannelManager implements Closeable {
+public class SpillChannelManager {
 
     private final HashSet<FileIOChannel.ID> channels;
     private final HashSet<FileIOChannel> openChannels;
 
-    private volatile boolean closed;
-
     public SpillChannelManager() {
         this.channels = new HashSet<>(64);
         this.openChannels = new HashSet<>(64);
     }
 
     /** Add a new File channel. */
     public synchronized void addChannel(FileIOChannel.ID id) {
-        checkArgument(!closed);
         channels.add(id);
     }
 
     /** Open File channels. */
     public synchronized void addOpenChannels(List<FileIOChannel> toOpen) {
-        checkArgument(!closed);
         for (FileIOChannel channel : toOpen) {
             openChannels.add(channel);
             channels.remove(channel.getChannelID());
         }
     }
 
     public synchronized void removeChannel(FileIOChannel.ID id) {
-        checkArgument(!closed);
         channels.remove(id);
     }
 
-    @Override
-    public synchronized void close() {
-
-        if (this.closed) {
-            return;
-        }
-
-        this.closed = true;
-
+    public synchronized void reset() {
         for (Iterator<FileIOChannel> channels = this.openChannels.iterator();
                 channels.hasNext(); ) {
             final FileIOChannel channel = channels.next();

paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java

@@ -188,8 +188,15 @@ public void testSpilling() throws Exception {
         innerTestSpilling(createBuffer());
     }
 
+    @Test
+    public void testSpillingAndClearWithMaxFanIn() throws Exception {
+        BinaryExternalSortBuffer buffer = createBuffer(2);
+        innerTestSpilling(buffer);
+        innerTestSpilling(buffer);
+    }
+
     private void innerTestSpilling(BinaryExternalSortBuffer sorter) throws Exception {
-        int size = 1000_000;
+        int size = 2000_000;
 
         MockBinaryRowReader reader = new MockBinaryRowReader(size);
         sorter.write(reader);