diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java index 3114baad2e83..4f2a23ef1a2a 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java @@ -38,6 +38,7 @@ import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Options; +import org.apache.hadoop.io.IOUtils; import java.io.IOException; import java.io.OutputStreamWriter; @@ -308,9 +309,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopFileIOSkipFullyTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopFileIOSkipFullyTest.java new file mode 100644 index 000000000000..3acb94f935fe --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopFileIOSkipFullyTest.java @@ -0,0 +1,97 @@ +/* + * 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.fs.hadoop; + +import org.apache.hadoop.fs.FSDataInputStream; +import org.junit.jupiter.api.Test; +import org.mockito.InOrder; + +import java.io.EOFException; +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; + +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +/** + * {@code HadoopSeekableInputStream#skipFully} turns a short forward seek into skips. A stream that + * returns 0 from {@code skip} used to spin forever; a 0 has to be resolved by reading, because + * {@link java.io.InputStream#skip} may return it without being at the end. + */ +class HadoopFileIOSkipFullyTest { + + @Test + void skipFullyThrowsWhenTheStreamReallyEnds() throws Exception { + FSDataInputStream in = mock(FSDataInputStream.class); + // a caller that reads a 0 as no progress asks again, forever. Fail on the second call so + // this test reports that rather than hanging the fork, which has no timeout to save it. + when(in.skip(anyLong())) + .thenReturn(0L) + .thenThrow(new AssertionError("skip was called again after returning 0")); + // the read probe is what distinguishes EOF from a transient zero + when(in.read()).thenReturn(-1); + + assertThatThrownBy(() -> skipFully(in, 4096L)).hasRootCauseInstanceOf(EOFException.class); + verify(in).read(); + } + + @Test + void skipFullyContinuesAfterATransientZero() throws Exception { + FSDataInputStream in = mock(FSDataInputStream.class); + // 0 first, then progress. The fail-fast revision threw here; the loop before it did not + // probe at all, so the read is what pins this case. + when(in.skip(anyLong())).thenReturn(0L, 4095L); + when(in.read()).thenReturn(7); + + assertThatCode(() -> skipFully(in, 4096L)).doesNotThrowAnyException(); + // the probe consumed one byte, so the second skip asks for the remaining 4095, and that + // is the whole conversation: an in-order verify alone would allow extra probes + InOrder inOrder = inOrder(in); + inOrder.verify(in).skip(4096L); + inOrder.verify(in).read(); + inOrder.verify(in).skip(4095L); + verifyNoMoreInteractions(in); + } + + @Test + void skipFullyIsANoOpForNothingToSkip() throws Exception { + FSDataInputStream in = mock(FSDataInputStream.class); + + assertThatCode(() -> skipFully(in, 0L)).doesNotThrowAnyException(); + verify(in, never()).skip(anyLong()); + } + + private static void skipFully(FSDataInputStream in, long bytes) throws Exception { + Class clazz = + Class.forName("org.apache.paimon.fs.hadoop.HadoopFileIO$HadoopSeekableInputStream"); + Constructor constructor = clazz.getDeclaredConstructor(FSDataInputStream.class); + constructor.setAccessible(true); + Object stream = constructor.newInstance(in); + Method skipFully = clazz.getDeclaredMethod("skipFully", long.class); + skipFully.setAccessible(true); + skipFully.invoke(stream, bytes); + } +} diff --git a/paimon-filesystems/paimon-azure-impl/src/main/java/org/apache/paimon/azure/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-azure-impl/src/main/java/org/apache/paimon/azure/HadoopCompliantFileIO.java index f928bde84b9b..2642a9d3fbae 100644 --- a/paimon-filesystems/paimon-azure-impl/src/main/java/org/apache/paimon/azure/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-azure-impl/src/main/java/org/apache/paimon/azure/HadoopCompliantFileIO.java @@ -27,6 +27,7 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.io.IOUtils; import javax.annotation.Nullable; @@ -214,9 +215,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } } diff --git a/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/HadoopCompliantFileIO.java index 36e9a1e829e4..06450d9f1baa 100644 --- a/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/HadoopCompliantFileIO.java @@ -27,6 +27,7 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.io.IOUtils; import javax.annotation.Nullable; @@ -214,9 +215,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } } diff --git a/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/HadoopCompliantFileIO.java index 227cddcedda7..0919408f6194 100644 --- a/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/HadoopCompliantFileIO.java @@ -27,6 +27,7 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.io.IOUtils; import java.io.IOException; import java.util.Map; @@ -214,9 +215,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } } diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java index 0ba8ef98d1b7..64fc3e07c708 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java @@ -33,6 +33,7 @@ import com.aliyun.jindodata.common.JindoHadoopSystem; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.io.IOUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -348,9 +349,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } } diff --git a/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/HadoopCompliantFileIO.java index 199b2ee2dd60..bbdbc339af26 100644 --- a/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/HadoopCompliantFileIO.java @@ -28,6 +28,7 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.io.IOUtils; import javax.annotation.Nullable; @@ -235,9 +236,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } } diff --git a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java index ab48da87cdfa..c3d794fae848 100644 --- a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java @@ -28,6 +28,7 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.io.IOUtils; import java.io.IOException; import java.io.UncheckedIOException; @@ -243,9 +244,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } } diff --git a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java index a662e8a07592..5965d7e6704f 100644 --- a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java @@ -28,6 +28,7 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.io.IOUtils; import java.io.IOException; import java.util.Map; @@ -235,9 +236,9 @@ public void forceSeek(long seekPos) throws IOException { * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { - while (bytes > 0) { - bytes -= in.skip(bytes); - } + // hadoop's helper probes with read() before calling it EOF, because skip may return 0 + // without being at the end. The loop this replaces subtracted that 0 and asked again. + IOUtils.skipFully(in, bytes); } }