From 8fe78607a6022a0501484f5bceed0cd69a117971 Mon Sep 17 00:00:00 2001 From: jackylee-ch Date: Thu, 3 Sep 2026 19:31:57 +0800 Subject: [PATCH] [common] Fail on a short remote read instead of caching zero padding The block returned by readRemote is cached, so tolerating a short read turned one truncated remote read into a durable zero-padded block. The VectoredReadable branch in the same method already throws via preadFully; use IOUtils.readFully for the other one. --- .../fs/cache/CachingSeekableInputStream.java | 21 ++++----------- .../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 48032c8a70e5..447912330fbf 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 @@ private byte[] readRemote(long offset, int size) throws IOException { } 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 @@ private void checkNotClosed() throws IOException { } } - 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 35e9ec7668c8..ded23667a972 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 @@ private CachingFileIO newCachingFileIO( 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 @@ private static class MockFileIO implements FileIO { new ConcurrentHashMap<>(); private final Map files = new HashMap<>(); + private final Map reportedLengths = new HashMap<>(); // concurrent so the thread-safety tests below can count from several reader threads private final Map fileStatusCalls = new ConcurrentHashMap<>(); private final Map newInputStreamCalls = new ConcurrentHashMap<>(); @@ -1024,6 +1042,12 @@ void addFile(String name, byte[] data) { 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 @@ public FileStatus getFileStatus(Path path) throws IOException { return new FileStatus() { @Override public long getLen() { - return data.length; + return reportedLengths.getOrDefault(name, (long) data.length); } @Override