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 a04735810a [fs] Delegate skipFully to hadoop's IOUtils (#9779)
a04735810a is described below
commit a04735810ac161986d2e7d0623d51cfbad14f544
Author: YangJie <[email protected]>
AuthorDate: Sun Sep 13 22:42:16 2026 -0400
[fs] Delegate skipFully to hadoop's IOUtils (#9779)
---
.../org/apache/paimon/fs/hadoop/HadoopFileIO.java | 7 +-
.../fs/hadoop/HadoopFileIOSkipFullyTest.java | 97 ++++++++++++++++++++++
.../apache/paimon/azure/HadoopCompliantFileIO.java | 7 +-
.../apache/paimon/cosn/HadoopCompliantFileIO.java | 7 +-
.../apache/paimon/gs/HadoopCompliantFileIO.java | 7 +-
.../apache/paimon/jindo/HadoopCompliantFileIO.java | 7 +-
.../apache/paimon/obs/HadoopCompliantFileIO.java | 7 +-
.../apache/paimon/oss/HadoopCompliantFileIO.java | 7 +-
.../apache/paimon/s3/HadoopCompliantFileIO.java | 7 +-
9 files changed, 129 insertions(+), 24 deletions(-)
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 3114baad2e..4f2a23ef1a 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.FSDataInputStream;
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 class HadoopFileIO implements FileIO,
HadoopOptionsProvider {
* @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 0000000000..3acb94f935
--- /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 f928bde84b..2642a9d3fb 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.paimon.fs.SeekableInputStream;
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 abstract class HadoopCompliantFileIO implements
FileIO {
* @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 36e9a1e829..06450d9f1b 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.paimon.fs.SeekableInputStream;
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 abstract class HadoopCompliantFileIO implements
FileIO {
* @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 227cddcedd..0919408f61 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.paimon.fs.SeekableInputStream;
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 abstract class HadoopCompliantFileIO implements
FileIO {
* @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 0ba8ef98d1..64fc3e07c7 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
org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
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 abstract class HadoopCompliantFileIO implements
FileIO {
* @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 199b2ee2dd..bbdbc339af 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.paimon.fs.SeekableInputStream;
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 abstract class HadoopCompliantFileIO implements
FileIO {
* @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 ab48da87cd..c3d794fae8 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.paimon.fs.SeekableInputStream;
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 abstract class HadoopCompliantFileIO implements
FileIO {
* @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 a662e8a075..5965d7e670 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.paimon.fs.SeekableInputStream;
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 abstract class HadoopCompliantFileIO implements
FileIO {
* @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);
}
}
