@@ -693,14 +693,22 @@ fn encode_block(key: &BlockKey, payload: &Bytes) -> Vec<u8> {
 }
 
 fn decode_block(key: &BlockKey, encoded: &[u8]) -> Result<Bytes, BlockDecodeError> {
-    let (decoded_key, payload) = decode_block_any(encoded)?;
+    let (decoded_key, payload) = validate_block(encoded)?;
     if &decoded_key != key {
         return Err(BlockDecodeError("cache block key does not match"));
     }
-    Ok(payload)
+    Ok(Bytes::copy_from_slice(&encoded[payload]))
 }
 
-fn decode_block_any(encoded: &[u8]) -> Result<(BlockKey, Bytes), BlockDecodeError> {
+fn decode_block_owned(key: &BlockKey, encoded: Vec<u8>) -> Result<Bytes, BlockDecodeError> {
+    let (decoded_key, payload) = validate_block(&encoded)?;
+    if &decoded_key != key {
+        return Err(BlockDecodeError("cache block key does not match"));
+    }
+    Ok(Bytes::from(encoded).slice(payload))
+}
+
+fn validate_block(encoded: &[u8]) -> Result<(BlockKey, std::ops::Range<usize>), BlockDecodeError> {
     let (key, layout) = decode_block_header(encoded, encoded.len() as u64)?;
     let payload_len = usize::try_from(layout.payload_len)
         .map_err(|_| BlockDecodeError("cache block payload is too large"))?;
@@ -717,10 +725,7 @@ fn decode_block_any(encoded: &[u8]) -> Result<(BlockKey, Bytes), BlockDecodeErro
         return Err(BlockDecodeError("cache block checksum does not match"));
     }
 
-    Ok((
-        key,
-        Bytes::copy_from_slice(&encoded[layout.header_len..payload_end]),
-    ))
+    Ok((key, layout.header_len..payload_end))
 }
 
 #[derive(Clone, Copy, Debug)]
@@ -846,6 +851,31 @@ mod tests {
         assert_eq!(decoded, payload);
     }
 
+    #[test]
+    fn test_owned_block_decode_reuses_allocation_and_validates() {
+        let key = BlockKey::new("s3://bucket/table/data.parquet", 1024, 3);
+        let payload = Bytes::from_static(b"cached block");
+        let encoded = encode_block(&key, &payload);
+        let (_, range) = validate_block(&encoded).unwrap();
+        let expected_address = encoded.as_ptr() as usize + range.start;
+        let decoded = decode_block_owned(&key, encoded).unwrap();
+        assert_eq!(decoded, payload);
+        assert_eq!(decoded.as_ptr() as usize, expected_address);
+
+        let wrong_key = BlockKey::new("s3://bucket/table/data.parquet", 1024, 4);
+        assert!(decode_block_owned(&wrong_key, encode_block(&key, &payload)).is_err());
+        let encoded = encode_block(&key, &payload);
+        for end in 0..encoded.len() {
+            assert!(decode_block_owned(&key, encoded[..end].to_vec()).is_err());
+        }
+        let mut corrupt = encoded;
+        corrupt[range.start] ^= 1;
+        assert!(decode_block_owned(&key, corrupt).is_err());
+        assert!(decode_block_owned(&key, encode_block(&key, &Bytes::new()))
+            .unwrap()
+            .is_empty());
+    }
+
     #[test]
     fn test_disk_block_codec_rejects_crc_corruption() {
         let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 1024, 3);
 
@@ -48,6 +48,8 @@ pub(crate) struct LocalCache {
     block_size: u64,
     whitelist: HashSet<FileType>,
     file_size_capacity: usize,
+    // Bound queued and active background fills.
+    fill_slots: Arc<tokio::sync::Semaphore>,
 }
 
 #[derive(Debug)]
@@ -76,6 +78,7 @@ impl LocalCache {
             block_size,
             whitelist: FileType::parse_whitelist(whitelist),
             file_size_capacity: DEFAULT_FILE_SIZE_CAPACITY,
+            fill_slots: Arc::new(tokio::sync::Semaphore::new(8)),
         })
     }
 
@@ -124,13 +127,30 @@ impl LocalCache {
             block_size: config.block_size,
             whitelist: config.whitelist,
             file_size_capacity,
+            fill_slots: Arc::new(tokio::sync::Semaphore::new(8)),
         })
     }
 
     fn block_size(&self) -> u64 {
         self.block_size
     }
 
+    fn background_fill_slot(&self) -> Option<tokio::sync::OwnedSemaphorePermit> {
+        // Current-thread runtimes stop driving spawned fills when block_on returns.
+        // Keep those fills synchronous so their block locks cannot outlive the read.
+        if !matches!(
+            tokio::runtime::Handle::try_current().map(|handle| handle.runtime_flavor()),
+            Ok(tokio::runtime::RuntimeFlavor::MultiThread)
+        ) {
+            return None;
+        }
+        // Limit background payloads to 8 MiB per cache; otherwise fill synchronously.
+        if !matches!(self.backend, CacheBackend::SharedDisk(_)) || self.block_size > 1024 * 1024 {
+            return None;
+        }
+        self.fill_slots.clone().try_acquire_owned().ok()
+    }
+
     fn block_key(&self, path: &str, block_index: u64) -> BlockKey {
         BlockKey::with_namespace(&self.namespace, path, self.block_size, block_index)
     }
 
@@ -82,8 +82,8 @@ impl CachedFileReader {
         }
 
         let load_lock = self.cache.block_load_lock(&key).await;
-        let load_guard = load_lock.lock().await;
-        let _shared_guard = self.cache.shared_load_lock(&key).await;
+        let load_guard = load_lock.clone().lock_owned().await;
+        let shared_guard = self.cache.shared_load_lock(&key).await;
         let result = if let Some(payload) = self
             .cache
             .get_block(&key, expected_len, &self.read_token)
@@ -107,6 +107,20 @@ impl CachedFileReader {
         } else {
             match self.delegate.read(start..end).await {
                 Ok(payload) if payload.len() == expected_len => {
+                    if let Some(slot) = self.cache.background_fill_slot() {
+                        let cache = self.cache.clone();
+                        let token = self.read_token.clone();
+                        let fill_payload = payload.clone();
+                        // Hold download guards until publication completes.
+                        tokio::spawn(async move {
+                            let _slot = slot;
+                            let _shared_guard = shared_guard;
+                            cache.put_block(&key, fill_payload, &token).await;
+                            drop(load_guard);
+                            cache.release_block_load_lock(&key, &load_lock).await;
+                        });
+                        return Ok(payload);
+                    }
                     self.cache
                         .put_block(&key, payload.clone(), &self.read_token)
                         .await;
@@ -125,6 +139,7 @@ impl CachedFileReader {
                 Err(error) => Err(error),
             }
         };
+        drop(shared_guard);
         drop(load_guard);
         self.cache.release_block_load_lock(&key, &load_lock).await;
         result
@@ -356,6 +371,104 @@ mod tests {
         }
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn shared_fill_returns_before_publication_and_preserves_single_flight() {
+        let directory = tempfile::tempdir().unwrap();
+        let delegate = Arc::new(CountingReader {
+            data: Bytes::from_static(b"abcdefgh"),
+            reads: AtomicUsize::new(0),
+        });
+        let cache = Arc::new(
+            LocalCache::new(LocalCacheConfig {
+                shared: true,
+                dir: Some(directory.path().to_path_buf()),
+                namespace: "test".into(),
+                max_size: Some(4096),
+                block_size: 4,
+                whitelist: std::collections::HashSet::from([FileType::Meta]),
+            })
+            .unwrap(),
+        );
+        let reader = CachedFileReader::new(delegate.clone(), "data.parquet", 8, cache.clone());
+        let gate = std::fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(
+                directory
+                    .path()
+                    .join("paimon-shared-local-cache-v1/directory.lock"),
+            )
+            .unwrap();
+        gate.lock().unwrap();
+        assert_eq!(
+            tokio::time::timeout(std::time::Duration::from_millis(500), reader.read(0..4))
+                .await
+                .unwrap()
+                .unwrap(),
+            "abcd"
+        );
+        assert_eq!(cache.fill_slots.available_permits(), 7);
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(50), reader.read(0..4))
+                .await
+                .is_err()
+        );
+        assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+        drop(gate);
+        assert_eq!(reader.read(0..4).await.unwrap(), "abcd");
+        assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+        let permits = cache
+            .fill_slots
+            .clone()
+            .acquire_many_owned(8)
+            .await
+            .unwrap();
+        assert!(cache.background_fill_slot().is_none());
+        drop(permits);
+    }
+
+    #[test]
+    fn shared_fill_does_not_block_reads_on_another_runtime() {
+        let directory = tempfile::tempdir().unwrap();
+        let delegate = Arc::new(CountingReader {
+            data: Bytes::from_static(b"abcdefgh"),
+            reads: AtomicUsize::new(0),
+        });
+        let cache = Arc::new(
+            LocalCache::new(LocalCacheConfig {
+                shared: true,
+                dir: Some(directory.path().to_path_buf()),
+                namespace: "test".into(),
+                max_size: Some(4096),
+                block_size: 4,
+                whitelist: std::collections::HashSet::from([FileType::Meta]),
+            })
+            .unwrap(),
+        );
+        let reader = CachedFileReader::new(delegate.clone(), "data.parquet", 8, cache);
+        let first_runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        assert_eq!(first_runtime.block_on(reader.read(0..4)).unwrap(), "abcd");
+
+        // Keep the first runtime alive without driving its background tasks.
+        let second_runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let second = second_runtime.block_on(async {
+            tokio::time::timeout(std::time::Duration::from_millis(500), reader.read(0..4)).await
+        });
+        assert_eq!(
+            second
+                .expect("read must not wait for an idle runtime")
+                .unwrap(),
+            "abcd"
+        );
+        assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+    }
+
     #[tokio::test]
     async fn test_cached_range_single_flight_reads_cold_block_once() {
         let delegate = Arc::new(SlowCountingReader {