[Stephen0421]

Purpose

The local disk cache keys blocks by path and index only:

# pypaimon/filesystem/caching_file_io.py
key = f"{file_path}:{block_index}"

But a block's contents are sized by local-cache.block-size. Since the cache directory outlives the process, reopening it after retuning block-size matches blocks that were split at the previous size, and reads return bytes from the wrong file offset.

This is reachable through ordinary configuration tuning: the local cache docs show local-cache.dir and local-cache.block-size set together in the same example, the cache directory is never self-cleaned, and the default whitelist (meta,global-index) covers manifest, snapshot and schema files that feed query planning.

Reading a real Parquet table after changing only local-cache.block-size, reusing the same cache directory:

     64 b -> 256 b   | ArrowInvalid: Parquet magic bytes not found in footer
    256 b -> 64 b    | ArrowInvalid: Parquet magic bytes not found in footer
     1 kb -> 4 kb    | ArrowInvalid
     4 kb -> 1 kb    | ArrowInvalid
     64 b -> 64 b    | OK   (control)

At the stream level the corruption is silent rather than raising: a read_at(512, 512) can return bytes taken from offset 40, and in one direction it returns more bytes than requested, breaking the read_at contract.

Java already does this correctly — LocalDiskCacheManager.cachePath builds blockSize + ":" + filePath + ":" + blockIndex. This change aligns PyPaimon with the Java implementation:

key = f"{self._block_size}:{file_path}:{block_index}"

No migration is needed. Stale blocks simply stop matching, so a retuned cache starts cold instead of returning wrong data, and the existing LRU reclaims the old entries.

Note this does not address cache isolation between catalogs sharing one cache directory. Java scopes keys with a per-catalog cacheNamespace (CachingFileIO), which PyPaimon also lacks; that touches cache manager lifecycle and is better handled separately.

Tests

Three regression tests, at the cache manager, input stream and end-to-end Parquet read levels:

• LocalDiskCacheManagerTest.test_restart_with_changed_block_size_does_not_reuse_blocks
• CachingInputStreamTest.test_reads_stay_correct_after_block_size_change
• test_block_size_change_reuses_disk_cache_dir_safely (parametrized both directions)

All three fail without the fix and pass with it.

The existing disk cache tests all pin block_size=64 and never reuse a directory across sizes, which is why this went unnoticed — 76 cache tests passed on the unfixed code.