From fc26d7961cfcd8a55ae6d756e3f905b0712252e8 Mon Sep 17 00:00:00 2001 From: "wenchao.wu" Date: Thu, 29 Oct 2026 17:14:36 +0800 Subject: [PATCH] [python] Include block size in the local disk cache key The local disk cache keys blocks by path and index only, but block contents are sized by local-cache.block-size. Because the cache directory outlives the process, reopening it after retuning block-size matches blocks that were split at the previous size, so reads return bytes from the wrong file offset. Java's LocalDiskCacheManager already includes the block size in its key; this aligns PyPaimon with it. Stale blocks are no longer matched after the change, so a retuned cache starts cold and the existing LRU reclaims the old entries. Add regression coverage at the cache manager, input stream and end-to-end Parquet read levels. The existing disk cache tests all pin block_size=64, which is why this went unnoticed. --- .../pypaimon/filesystem/caching_file_io.py | 2 +- .../pypaimon/tests/caching_file_io_test.py | 32 +++++++++++++++++++ .../tests/parquet_local_cache_test.py | 28 ++++++++++++++++ 3 files changed, 61 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py b/paimon-python/pypaimon/filesystem/caching_file_io.py index d60d8c2ecdcf..3be2428bd824 100644 --- a/paimon-python/pypaimon/filesystem/caching_file_io.py +++ b/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) diff --git a/paimon-python/pypaimon/tests/caching_file_io_test.py b/paimon-python/pypaimon/tests/caching_file_io_test.py index ae5ff0704423..2391ef401159 100644 --- a/paimon-python/pypaimon/tests/caching_file_io_test.py +++ b/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) diff --git a/paimon-python/pypaimon/tests/parquet_local_cache_test.py b/paimon-python/pypaimon/tests/parquet_local_cache_test.py index 5779941008ba..f51c6dc7e0ed 100644 --- a/paimon-python/pypaimon/tests/parquet_local_cache_test.py +++ b/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()