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 527f2eecec [core] Fix invalid length when reading spilled data (#9183)
527f2eecec is described below
commit 527f2eececb3d0e0e72f3374da62fd57e7e8e65b
Author: tsreaper <[email protected]>
AuthorDate: Wed Aug 12 14:05:33 2026 +0800
[core] Fix invalid length when reading spilled data (#9183)
---
.../paimon/disk/BufferFileChannelReader.java | 12 +++-
.../paimon/disk/ChannelWriterOutputView.java | 4 +-
.../paimon/disk/ChannelWriterOutputViewTest.java | 69 ++++++++++++++++++++++
3 files changed, 82 insertions(+), 3 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/disk/BufferFileChannelReader.java
b/paimon-core/src/main/java/org/apache/paimon/disk/BufferFileChannelReader.java
index 6762edeb40..15520b59ef 100644
---
a/paimon-core/src/main/java/org/apache/paimon/disk/BufferFileChannelReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/disk/BufferFileChannelReader.java
@@ -46,7 +46,7 @@ public class BufferFileChannelReader {
// Read header
header.clear();
- fileChannel.read(header);
+ readFully(header);
header.flip();
int size = header.getInt();
@@ -60,8 +60,16 @@ public class BufferFileChannelReader {
}
checkArgument(buffer.getSize() == 0, "Buffer not empty");
- fileChannel.read(buffer.getNioBuffer(0, size));
+ readFully(buffer.getNioBuffer(0, size));
buffer.setSize(size);
return fileChannel.size() - fileChannel.position() == 0;
}
+
+ private void readFully(ByteBuffer target) throws IOException {
+ while (target.hasRemaining()) {
+ if (fileChannel.read(target) < 0) {
+ throw new IOException("Premature end of file while reading
buffer");
+ }
+ }
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/disk/ChannelWriterOutputView.java
b/paimon-core/src/main/java/org/apache/paimon/disk/ChannelWriterOutputView.java
index c2d3cc2051..3f759cdcca 100644
---
a/paimon-core/src/main/java/org/apache/paimon/disk/ChannelWriterOutputView.java
+++
b/paimon-core/src/main/java/org/apache/paimon/disk/ChannelWriterOutputView.java
@@ -65,7 +65,9 @@ public final class ChannelWriterOutputView extends
AbstractPagedOutputView imple
public void close() throws IOException {
if (!writer.isClosed()) {
int currentPositionInSegment = getCurrentPositionInSegment();
- writeCompressed(currentSegment, currentPositionInSegment);
+ if (currentPositionInSegment > 0) {
+ writeCompressed(currentSegment, currentPositionInSegment);
+ }
clear();
this.writeBytes = writer.getSize();
this.writer.close();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/disk/ChannelWriterOutputViewTest.java
b/paimon-core/src/test/java/org/apache/paimon/disk/ChannelWriterOutputViewTest.java
new file mode 100644
index 0000000000..9ad0c562e1
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/disk/ChannelWriterOutputViewTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.disk;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.BlockCompressionType;
+import org.apache.paimon.data.serializer.BinaryRowSerializer;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ChannelWriterOutputView}. */
+public class ChannelWriterOutputViewTest {
+
+ private static final int BLOCK_SIZE = 64 * 1024;
+
+ @TempDir Path tempDir;
+
+ @Test
+ public void testEmptyOutput() throws Exception {
+ BlockCompressionFactory compressionFactory =
+ BlockCompressionFactory.create(BlockCompressionType.LZ4);
+ try (IOManager ioManager = IOManager.create(tempDir.toString())) {
+ FileIOChannel.ID channel = ioManager.createChannel();
+ ChannelWriterOutputView output =
+ FileChannelUtil.createOutputView(
+ ioManager, channel, compressionFactory,
BLOCK_SIZE);
+ output.close();
+
+ ChannelReaderInputView input =
+ new ChannelReaderInputView(
+ channel,
+ ioManager,
+ compressionFactory,
+ BLOCK_SIZE,
+ output.getBlockCount());
+ try {
+ assertThat(
+ new ChannelReaderInputViewIterator(
+ input, null, new
BinaryRowSerializer(1))
+ .next())
+ .isNull();
+ assertThat(output.getBlockCount()).isZero();
+ } finally {
+ input.getChannel().closeAndDelete();
+ }
+ }
+ }
+}
