[python] Include block size in the local disk cache key

paimon-python/pypaimon/filesystem/caching_file_io.py

@@ -170,7 +170,7 @@ def block_size(self) -> int:
         return self._block_size
 
     def _cache_path(self, file_path: str, block_index: int) -> str:
-        key = f"{file_path}:{block_index}"
+        key = f"{self._block_size}:{file_path}:{block_index}"
         h = hashlib.sha256(key.encode('utf-8')).hexdigest()
         prefix = h[:2]
         sub_dir = os.path.join(self._cache_dir, prefix)

paimon-python/pypaimon/tests/caching_file_io_test.py

@@ -137,6 +137,20 @@ def test_scan_size_on_restart(self):
         self.assertEqual(b"x" * 100, cache2.get_block("f", 0))
         self.assertEqual(b"y" * 200, cache2.get_block("f", 1))
 
+    def test_restart_with_changed_block_size_does_not_reuse_blocks(self):
+        # Blocks are sized by block_size, so a directory written with one size
+        # must never satisfy reads made with another.
+        cache1 = LocalDiskCacheManager(self.cache_dir, 2 ** 63 - 1, block_size=64)
+        cache1.put_block("f", 0, b"x" * 64)
+
+        for block_size in (32, 128, 256):
+            cache2 = LocalDiskCacheManager(self.cache_dir, 2 ** 63 - 1, block_size=block_size)
+            self.assertIsNone(cache2.get_block("f", 0))
+
+        # The original block stays readable for a manager using its own size.
+        cache3 = LocalDiskCacheManager(self.cache_dir, 2 ** 63 - 1, block_size=64)
+        self.assertEqual(b"x" * 64, cache3.get_block("f", 0))
+
     def test_cache_dir_created(self):
         new_dir = os.path.join(self.cache_dir, "sub", "deep")
         LocalDiskCacheManager(new_dir, 2 ** 63 - 1, block_size=64)
@@ -307,6 +321,24 @@ def test_partial_last_block(self):
         # Verify block 1 is only 2 bytes
         self.assertEqual(b"ij", cache.get_block("test", 1))
 
+    def test_reads_stay_correct_after_block_size_change(self):
+        # A disk cache directory outlives the process, so reopening it with a
+        # different block-size must re-fetch rather than reuse stale blocks.
+        data = bytes(i % 251 for i in range(4096))
+        for first, second in [(64, 256), (256, 64), (128, 512)]:
+            with self.subTest(first=first, second=second):
+                cache_dir = tempfile.mkdtemp()
+                try:
+                    warm = LocalDiskCacheManager(cache_dir, 2 ** 63 - 1, block_size=first)
+                    CachingInputStream(self._make_file_io(data), "test", warm).read()
+
+                    cache = LocalDiskCacheManager(cache_dir, 2 ** 63 - 1, block_size=second)
+                    stream = CachingInputStream(self._make_file_io(data), "test", cache)
+                    self.assertEqual(data, stream.read())
+                    self.assertEqual(data[512:1024], stream.read_at(512, 512))
+                finally:
+                    shutil.rmtree(cache_dir, ignore_errors=True)
+
     def test_coalesce_cold_blocks_and_preserve_cursor(self):
         data = b"abcdefghijklmn"
         cache = LocalMemoryCacheManager(1024, block_size=4)

paimon-python/pypaimon/tests/parquet_local_cache_test.py

@@ -215,3 +215,31 @@ def read():
         assert cold_opens > 0
         assert read() == [1, 2, 3]
         assert opens.call_count == cold_opens
+
+
+@pytest.mark.parametrize('first,second', [('64 b', '256 b'), ('256 b', '64 b')])
+def test_block_size_change_reuses_disk_cache_dir_safely(tmp_path, first, second):
+    # A disk cache directory survives the process, so a catalog that reopens it
+    # with a retuned local-cache.block-size must still read the table correctly.
+    path = tmp_path / 'data.parquet'
+    expected = list(range(1, 501))
+    pq.write_table(pa.table({'v': pa.array(expected, type=pa.int32())}),
+                   path, row_group_size=100)
+    cache_dir = str(tmp_path / 'cache')
+
+    def read_with(block_size):
+        opts = Options({'local-cache.enabled': 'true',
+                        'local-cache.dir': cache_dir,
+                        'local-cache.block-size': block_size,
+                        'local-cache.whitelist': 'structured-data'})
+        delegate = LocalFileIO(catalog_options=opts)
+        file_io = CachingFileIO.wrap_with_caching_if_needed(
+            delegate, opts, CachingFileIO.create_cache_manager(opts))
+        _reset_file_format_dataset_cache()
+        return _read(file_io, path.as_uri(), 'parquet', None)
+
+    try:
+        assert read_with(first) == expected
+        assert read_with(second) == expected
+    finally:
+        _reset_file_format_dataset_cache()