This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 9b5287881c [core] Reuse sidecar block cache for unfiltered scans
(#9924)
9b5287881c is described below
commit 9b5287881ca4c4319a847a585cb6cee436ba61ff
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
