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..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); @@ -402,8 +422,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/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 db914239eea5..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; @@ -1856,7 +1857,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 +1865,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 +1876,106 @@ 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); + 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