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 4242ff2fb9 [common] Fail on a short remote read instead of caching
zero padding (#9593)
4242ff2fb9 is described below
commit 4242ff2fb94d6d154dad540d14a81554756a1bb4
Author: jackylee <[email protected]>
AuthorDate: Fri Sep 4 15:21:39 2026 +0800
[common] Fail on a short remote read instead of caching zero padding (#9593)
---
.../fs/cache/CachingSeekableInputStream.java | 21 +++++------------
.../apache/paimon/fs/cache/CachingFileIOTest.java | 26 +++++++++++++++++++++-
2 files changed, 30 insertions(+), 17 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
index 48032c8a70..447912330f 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
@@ -195,7 +195,11 @@ public class CachingSeekableInputStream extends
SeekableInputStream implements V
}
synchronized (stream) {
stream.seek(offset);
- return readFully(stream, size);
+ // must not tolerate a short read: the block is handed to
putBlock, so a zero-padded
+ // tail would be cached and returned as file content for every
later read
+ byte[] buf = new byte[size];
+ IOUtils.readFully(stream, buf, 0, size);
+ return buf;
}
}
@@ -236,21 +240,6 @@ public class CachingSeekableInputStream extends
SeekableInputStream implements V
}
}
- private static byte[] readFully(SeekableInputStream in, int size) throws
IOException {
- byte[] buf = new byte[size];
- int remaining = size;
- int off = 0;
- while (remaining > 0) {
- int n = in.read(buf, off, remaining);
- if (n < 0) {
- break;
- }
- off += n;
- remaining -= n;
- }
- return buf;
- }
-
@Override
public void close() throws IOException {
// takes no lock, so it never waits behind an in-flight remote open
diff --git
a/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java
b/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java
index 35e9ec7668..ded23667a9 100644
---
a/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java
@@ -138,6 +138,23 @@ class CachingFileIOTest {
return new CachingFileIO(delegate, cache, whitelist);
}
+ @Test
+ void testShortRemoteReadIsNotCachedAsZeroPaddedBlock() throws IOException {
+ byte[] data = "truncated".getBytes();
+ MockFileIO delegate = new MockFileIO();
+ // the status says 8 bytes more than the stream can hand out
+ delegate.addTruncatedFile("snapshot-1", data, data.length + 8);
+
+ LocalDiskCacheManager cache = new LocalDiskCacheManager(cacheDir,
Long.MAX_VALUE, 64);
+ CachingFileIO cachingIO = newCachingFileIO(delegate, cache,
EnumSet.of(FileType.META), 64);
+
+ try (SeekableInputStream s = cachingIO.newInputStream(new
Path("snapshot-1"))) {
+ assertThatThrownBy(() -> readAll(s, data.length + 8))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Premature EOF");
+ }
+ }
+
@Test
void testMetaFileIsCached() throws IOException {
byte[] data = "snapshot data".getBytes();
@@ -976,6 +993,7 @@ class CachingFileIOTest {
new ConcurrentHashMap<>();
private final Map<String, byte[]> files = new HashMap<>();
+ private final Map<String, Long> reportedLengths = new HashMap<>();
// concurrent so the thread-safety tests below can count from several
reader threads
private final Map<String, Integer> fileStatusCalls = new
ConcurrentHashMap<>();
private final Map<String, Integer> newInputStreamCalls = new
ConcurrentHashMap<>();
@@ -1024,6 +1042,12 @@ class CachingFileIOTest {
files.put(name, data);
}
+ /** Reports a length beyond the bytes on hand, the way a truncated
remote file does. */
+ void addTruncatedFile(String name, byte[] data, long reportedLength) {
+ files.put(name, data);
+ reportedLengths.put(name, reportedLength);
+ }
+
int getFileStatusCallCount(String name) {
return fileStatusCalls.getOrDefault(name, 0);
}
@@ -1084,7 +1108,7 @@ class CachingFileIOTest {
return new FileStatus() {
@Override
public long getLen() {
- return data.length;
+ return reportedLengths.getOrDefault(name, (long)
data.length);
}
@Override
