From 00fbe9d9d612adbc16a26c5b14483123bb296965 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 17 Sep 2026 17:46:17 +0800 Subject: [PATCH 1/2] [core] Reuse sidecar block cache for unfiltered scans --- docs/docs/concepts/spec/manifest.md | 8 +- .../apache/paimon/manifest/ManifestFile.java | 4 +- .../paimon/manifest/ManifestFileTest.java | 93 ++++++++++++++++++- 3 files changed, 97 insertions(+), 8 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 5b4271cc215e..6f72db906743 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -77,9 +77,11 @@ writers and scans use sidecars when `manifest.sidecar.enabled` is true; when uns the completed output manifest and publish its `_EXTRA_FILES` reference only after both files close successfully. Failed writes and aborted writers clean up their own manifest/sidecar pairs. Scans with partition, row-ID or bucket filters select blocks before reading manifest entries. -Normal entry filtering and ADD/DELETE reconciliation still apply. Missing or unusable sidecars -fall back to normal manifest reads; disabled sidecars and scans without these filters do not -perform sidecar I/O. Sidecar caching is controlled by the catalog option +Unfiltered scans also select all sidecar blocks when a manifest cache is configured, so unfiltered +prefetches and later filtered reads share the same block cache. Without a manifest cache, unfiltered +scans keep the normal whole-manifest read path. Normal entry filtering and ADD/DELETE reconciliation +still apply. Missing or unusable sidecars fall back to normal manifest reads; disabled sidecars do +not perform sidecar I/O. Sidecar caching is controlled by the catalog option `cache.manifest-sidecar.max-memory` (64 MiB by default). A positive value supplies an additional budget independent of the manifest content cache. When set to 0, sidecars reuse the manifest content cache, or remain uncached if that cache is disabled. Sidecar caching diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java index 9740961e73d2..9968b05e7ee4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java @@ -402,8 +402,8 @@ public ManifestSidecar.Selection selectBlocks( @Nullable RowRangeIndex query, @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter) { - return !options.manifestSidecarEnabled() - || (query == null && partitionFilter == null && bucketFilter == null) + boolean hasFilter = query != null || partitionFilter != null || bucketFilter != null; + return !options.manifestSidecarEnabled() || (!hasFilter && cache == null) ? null : ManifestSidecar.read( fileIO, diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index db914239eea5..4e35dde4c694 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java @@ -1856,7 +1856,7 @@ void testPartitionOnlyPlanningUsesBlocksWithoutRowIds() { } @Test - void testUnknownRowIdKeepsPartitionIndexAndNoQueryDoesNotReadSidecar() { + void testUnfilteredReadWithoutCacheSkipsSidecar() { Options options = new Options(); options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); @@ -1864,7 +1864,8 @@ void testUnknownRowIdKeepsPartitionIndexAndNoQueryDoesNotReadSidecar() { ManifestFile manifests = createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, fileIO) .create(); - ManifestFileMeta meta = manifests.write(Collections.singletonList(gen.next())).get(0); + ManifestEntry entry = gen.next(); + ManifestFileMeta meta = manifests.write(Collections.singletonList(entry)).get(0); assertThat(ManifestSidecar.fileName(meta)).isNotNull(); assertThat( java.nio.file.Files.exists( @@ -1874,7 +1875,93 @@ void testUnknownRowIdKeepsPartitionIndexAndNoQueryDoesNotReadSidecar() { fileIO.reset(); assertThat(manifests.selectBlocks(meta, null)).isNull(); - assertThat(fileIO.opened).isEmpty(); + assertThat(manifests.read(meta.fileName())).containsExactly(entry); + assertThat(fileIO.opened) + .containsExactly(new Path(tempDir.toString(), "manifest/" + meta.fileName())); + } + + @Test + void testUnfilteredReadWithCacheAndWithoutSidecarUsesWholeManifestCache() { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO io = new RecordingFileIO(); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(16), Long.MAX_VALUE); + ManifestFile manifests = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io, cache) + .create(); + ManifestEntry entry = gen.next(); + ManifestFileMeta written = manifests.write(Collections.singletonList(entry)).get(0); + ManifestFileMeta unindexed = withExtraFiles(written, null); + Path manifestPath = new Path(tempDir.toString(), "manifest/" + written.fileName()); + + io.reset(); + ManifestSidecar.Selection selected = manifests.selectBlocks(unindexed, null); + assertThat(selected).isNull(); + assertThat( + manifests.read( + unindexed.fileName(), + unindexed.fileSize(), + null, + null, + row -> true, + manifestEntry -> true, + java.util.function.Function.identity(), + selected)) + .containsExactly(entry); + assertThat(io.opened).containsExactly(manifestPath); + assertThat(cache.getIfPresents(manifestPath)).isNotNull(); + + io.reset(); + assertThat(manifests.read(unindexed.fileName())).containsExactly(entry); + assertThat(io.opened).isEmpty(); + } + + @Test + void testUnfilteredReadWarmsBlockCacheForFilteredRead() throws Exception { + Options options = new Options(); + options.set(CoreOptions.BUCKET, 4); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO io = new RecordingFileIO(); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(16), Long.MAX_VALUE); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io, cache); + List entries = new ArrayList<>(); + for (int i = 0; i < 4000; i++) { + ManifestEntry entry = gen.next(); + entries.add( + ManifestEntry.create( + FileKind.ADD, entry.partition(), i / 1000, 4, entry.file())); + } + ManifestFileMeta meta = factory.create().write(entries).get(0); + ManifestFile manifests = factory.create(); + Path manifestPath = new Path(tempDir.toString(), "manifest/" + meta.fileName()); + Path sidecarPath = ManifestSidecar.path(manifestPath); + + io.reset(); + ManifestSidecar.Selection allBlocks = manifests.selectBlocks(meta, null); + assertThat(readSelectedEntries(manifests, meta, allBlocks)) + .containsExactlyElementsOf(entries); + assertThat(io.opened).containsExactly(sidecarPath, manifestPath); + assertThat(cache.getIfPresents(manifestPath)).isNull(); + + BucketFilter bucketFilter = new BucketFilter(false, 1, null, null); + io.reset(); + ManifestSidecar.Selection selected = manifests.selectBlocks(meta, null, null, bucketFilter); + assertThat( + manifests.read( + meta.fileName(), + meta.fileSize(), + null, + bucketFilter, + row -> true, + entry -> true, + java.util.function.Function.identity(), + selected)) + .containsExactlyElementsOf(entries.subList(1000, 2000)); + assertThat(io.opened).isEmpty(); + assertThat(io.bytes.get()).isZero(); } @Test From d3e178e29726b5f6ef1ad6d989dc7de684a09374 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 17 Sep 2026 18:08:07 +0800 Subject: [PATCH 2/2] [core] Preserve metrics for manifest block cache --- .../apache/paimon/manifest/ManifestFile.java | 46 ++++++++++++----- .../paimon/manifest/ManifestSidecar.java | 50 +++++++++++++++++-- .../paimon/manifest/ManifestFileTest.java | 16 +++++- 3 files changed, 95 insertions(+), 17 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java index 9968b05e7ee4..afc122c7235d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java @@ -63,6 +63,7 @@ public class ManifestFile extends ObjectsFile { private final long suggestedFileSize; private final CoreOptions options; @Nullable private final SegmentsCache sidecarCache; + @Nullable private CacheMetrics cacheMetrics; private ManifestFile( FileIO fileIO, @@ -116,6 +117,7 @@ protected ManifestEntryCache createCache( @Override public ManifestFile withCacheMetrics(@Nullable CacheMetrics cacheMetrics) { super.withCacheMetrics(cacheMetrics); + this.cacheMetrics = cacheMetrics; return this; } @@ -177,16 +179,32 @@ public List read( return cache.read(path, fileSize, filters, convertor); } - CloseableIterator iterator = - createManifestIterator( - fileIO, - path, - ManifestEntry.MANIFEST_ROW_TYPE, - partitionFilter, - bucketFilter, - selected, - cache == null ? null : cache.segmentsCache()); - return readFromIterator(iterator, serializer, readFilter, readTFilter, convertor); + CacheMetrics metrics = cacheMetrics; + ManifestSidecar.CacheStatus cacheStatus = + selected != null && cache != null && metrics != null + ? new ManifestSidecar.CacheStatus() + : null; + try { + CloseableIterator iterator = + createManifestIterator( + fileIO, + path, + ManifestEntry.MANIFEST_ROW_TYPE, + partitionFilter, + bucketFilter, + selected, + cache == null ? null : cache.segmentsCache(), + cacheStatus); + return readFromIterator(iterator, serializer, readFilter, readTFilter, convertor); + } finally { + if (cacheStatus != null) { + if (cacheStatus.hit()) { + metrics.increaseHitObject(); + } else { + metrics.increaseMissedObject(); + } + } + } } catch (IOException e) { throw new UncheckedIOException(e); } @@ -240,7 +258,7 @@ private static CloseableIterator createManifestIterator( @Nullable BucketFilter bucketFilter) throws IOException { return createManifestIterator( - fileIO, path, projectedType, partitionFilter, bucketFilter, null, null); + fileIO, path, projectedType, partitionFilter, bucketFilter, null, null, null); } private static CloseableIterator createManifestIterator( @@ -250,12 +268,14 @@ private static CloseableIterator createManifestIterator( @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter, @Nullable ManifestSidecar.Selection selected, - @Nullable SegmentsCache cache) + @Nullable SegmentsCache cache, + @Nullable ManifestSidecar.CacheStatus cacheStatus) throws IOException { try { ManifestAvroReader reader = new ManifestAvroReader( - ManifestSidecar.openManifest(fileIO, path, selected, cache)); + ManifestSidecar.openManifest( + fileIO, path, selected, cache, cacheStatus)); return reader.read(projectedType, partitionFilter, bucketFilter); } catch (IOException e) { FileUtils.checkExists(fileIO, path); diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java index 6a8ce2b143bb..f15dfb8d0722 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java @@ -692,7 +692,7 @@ private static byte[] readBytes(FileIO io, Path path) throws IOException { static InputStream openManifest(FileIO io, Path path, @Nullable Selection selected) throws IOException { - return openManifest(io, path, selected, null); + return openManifest(io, path, selected, null, null); } static InputStream openManifest( @@ -701,9 +701,38 @@ static InputStream openManifest( @Nullable Selection selected, @Nullable SegmentsCache cache) throws IOException { + return openManifest(io, path, selected, cache, null); + } + + static InputStream openManifest( + FileIO io, + Path path, + @Nullable Selection selected, + @Nullable SegmentsCache cache, + @Nullable CacheStatus cacheStatus) + throws IOException { return selected == null ? io.newInputStream(path) - : new SelectedBlockInput(io, path, selected, cache); + : new SelectedBlockInput(io, path, selected, cache, cacheStatus); + } + + /** Per-manifest status for reporting whether every selected block was served by the cache. */ + static final class CacheStatus { + private boolean accessed; + private boolean missed; + + private void hitBlock() { + accessed = true; + } + + private void miss() { + accessed = true; + missed = true; + } + + boolean hit() { + return accessed && !missed; + } } /** Separates physical byte ranges from whole-file cache keys. */ @@ -753,6 +782,7 @@ private static final class SelectedBlockInput extends InputStream { private final Path path; private final Selection selected; @Nullable private final SegmentsCache cache; + @Nullable private final CacheStatus cacheStatus; @Nullable private SeekableInputStream input; private boolean closed; private int headerPosition; @@ -763,11 +793,16 @@ private static final class SelectedBlockInput extends InputStream { private int bufferLimit; private SelectedBlockInput( - FileIO io, Path path, Selection selected, @Nullable SegmentsCache cache) { + FileIO io, + Path path, + Selection selected, + @Nullable SegmentsCache cache, + @Nullable CacheStatus cacheStatus) { this.io = io; this.path = path; this.selected = selected; this.cache = cache; + this.cacheStatus = cacheStatus; } @Override @@ -822,6 +857,9 @@ private boolean fillBuffer() throws IOException { > cache.maxElementSize())) { end += selected.blocks.get(blockPosition++).length; } + if (cacheStatus != null) { + cacheStatus.miss(); + } seekInput(block.offset); remaining = end - block.offset; } @@ -841,9 +879,15 @@ private boolean fillBuffer() throws IOException { private void readCachedBlocks(Block first) throws IOException { byte[] cached = cachedBlock(first); if (cached != null) { + if (cacheStatus != null) { + cacheStatus.hitBlock(); + } blockPosition++; buffer = cached; } else { + if (cacheStatus != null) { + cacheStatus.miss(); + } int firstPosition = blockPosition++; long end = first.offset + first.length; while (blockPosition < selected.blocks.size()) { diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index 4e35dde4c694..5172c305216b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java @@ -36,6 +36,7 @@ import org.apache.paimon.io.DataFileMetaWriteColsLegacySerializer; import org.apache.paimon.operation.AppendOnlyFileStoreScan; import org.apache.paimon.operation.ManifestsReader; +import org.apache.paimon.operation.metrics.CacheMetrics; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; @@ -1935,7 +1936,8 @@ void testUnfilteredReadWarmsBlockCacheForFilteredRead() throws Exception { FileKind.ADD, entry.partition(), i / 1000, 4, entry.file())); } ManifestFileMeta meta = factory.create().write(entries).get(0); - ManifestFile manifests = factory.create(); + CacheMetrics metrics = new CacheMetrics(); + ManifestFile manifests = factory.create().withCacheMetrics(metrics); Path manifestPath = new Path(tempDir.toString(), "manifest/" + meta.fileName()); Path sidecarPath = ManifestSidecar.path(manifestPath); @@ -1945,6 +1947,16 @@ void testUnfilteredReadWarmsBlockCacheForFilteredRead() throws Exception { .containsExactlyElementsOf(entries); assertThat(io.opened).containsExactly(sidecarPath, manifestPath); assertThat(cache.getIfPresents(manifestPath)).isNull(); + assertThat(metrics.getMissedObject()).hasValue(1); + assertThat(metrics.getHitObject()).hasValue(0); + + io.reset(); + ManifestSidecar.Selection cachedBlocks = manifests.selectBlocks(meta, null); + assertThat(readSelectedEntries(manifests, meta, cachedBlocks)) + .containsExactlyElementsOf(entries); + assertThat(io.opened).isEmpty(); + assertThat(metrics.getMissedObject()).hasValue(1); + assertThat(metrics.getHitObject()).hasValue(1); BucketFilter bucketFilter = new BucketFilter(false, 1, null, null); io.reset(); @@ -1962,6 +1974,8 @@ void testUnfilteredReadWarmsBlockCacheForFilteredRead() throws Exception { .containsExactlyElementsOf(entries.subList(1000, 2000)); assertThat(io.opened).isEmpty(); assertThat(io.bytes.get()).isZero(); + assertThat(metrics.getMissedObject()).hasValue(1); + assertThat(metrics.getHitObject()).hasValue(2); } @Test