This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 127c26452df0c4d757ebf17218dc039c301583a3 Author: Jingsong Lee <[email protected]> AuthorDate: Thu Sep 17 18:24:55 2026 +0800 [core] Reuse sidecar block cache for unfiltered scans (#9924) --- docs/docs/concepts/spec/manifest.md | 8 +- .../org/apache/paimon/manifest/ManifestFile.java | 50 +++++++--- .../apache/paimon/manifest/ManifestSidecar.java | 50 +++++++++- .../apache/paimon/manifest/ManifestFileTest.java | 107 ++++++++++++++++++++- 4 files changed, 191 insertions(+), 24 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 5b4271cc21..6f72db9067 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 9740961e73..afc122c723 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<ManifestEntry> { private final long suggestedFileSize; private final CoreOptions options; @Nullable private final SegmentsCache<Path> sidecarCache; + @Nullable private CacheMetrics cacheMetrics; private ManifestFile( FileIO fileIO, @@ -116,6 +117,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { @Override public ManifestFile withCacheMetrics(@Nullable CacheMetrics cacheMetrics) { super.withCacheMetrics(cacheMetrics); + this.cacheMetrics = cacheMetrics; return this; } @@ -177,16 +179,32 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { return cache.read(path, fileSize, filters, convertor); } - CloseableIterator<InternalRow> 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<InternalRow> 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 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { @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<InternalRow> createManifestIterator( @@ -250,12 +268,14 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter, @Nullable ManifestSidecar.Selection selected, - @Nullable SegmentsCache<Object> cache) + @Nullable SegmentsCache<Object> 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); @@ -402,8 +422,8 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { @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/main/java/org/apache/paimon/manifest/ManifestSidecar.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java index 6a8ce2b143..f15dfb8d07 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 @@ public final class ManifestSidecar { 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 @@ public final class ManifestSidecar { @Nullable Selection selected, @Nullable SegmentsCache<Object> cache) throws IOException { + return openManifest(io, path, selected, cache, null); + } + + static InputStream openManifest( + FileIO io, + Path path, + @Nullable Selection selected, + @Nullable SegmentsCache<Object> 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 @@ public final class ManifestSidecar { private final Path path; private final Selection selected; @Nullable private final SegmentsCache<Object> cache; + @Nullable private final CacheStatus cacheStatus; @Nullable private SeekableInputStream input; private boolean closed; private int headerPosition; @@ -763,11 +793,16 @@ public final class ManifestSidecar { private int bufferLimit; private SelectedBlockInput( - FileIO io, Path path, Selection selected, @Nullable SegmentsCache<Object> cache) { + FileIO io, + Path path, + Selection selected, + @Nullable SegmentsCache<Object> cache, + @Nullable CacheStatus cacheStatus) { this.io = io; this.path = path; this.selected = selected; this.cache = cache; + this.cacheStatus = cacheStatus; } @Override @@ -822,6 +857,9 @@ public final class ManifestSidecar { > cache.maxElementSize())) { end += selected.blocks.get(blockPosition++).length; } + if (cacheStatus != null) { + cacheStatus.miss(); + } seekInput(block.offset); remaining = end - block.offset; } @@ -841,9 +879,15 @@ public final class ManifestSidecar { 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 db914239ee..5172c30521 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.DataFileMeta; 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; @@ -1856,7 +1857,7 @@ public class ManifestFileTest { } @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 +1865,8 @@ public class ManifestFileTest { 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 +1876,106 @@ public class ManifestFileTest { 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<Path> 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<Path> cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(16), Long.MAX_VALUE); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io, cache); + List<ManifestEntry> 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); + CacheMetrics metrics = new CacheMetrics(); + ManifestFile manifests = factory.create().withCacheMetrics(metrics); + 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(); + 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(); + 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(); + assertThat(metrics.getMissedObject()).hasValue(1); + assertThat(metrics.getHitObject()).hasValue(2); } @Test
