[common] Fail on a short remote read instead of caching zero padding

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

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<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 @@ 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