From ac943bc09e3e1031b045e38c05607495e4a0ee47 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 16 Sep 2026 19:50:27 +0800 Subject: [PATCH 1/4] [core] Prune manifest blocks with sidecars during scans --- docs/docs/concepts/spec/manifest.md | 11 +- .../apache/paimon/manifest/BucketFilter.java | 16 + .../apache/paimon/manifest/ManifestFile.java | 74 ++- .../operation/AbstractFileStoreScan.java | 17 +- .../org/apache/paimon/utils/ObjectsCache.java | 6 + .../paimon/manifest/ManifestFileTest.java | 581 +++++++++++++++++- 6 files changed, 681 insertions(+), 24 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 29466c1ad683..d05a90e9a477 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -72,13 +72,16 @@ using independent partition, row-ID and bucket coverage. A sidecar uses the derived file name. The Avro schemas and `_VERSION` identifiers remain unchanged. The utility includes construction, validation, block selection and optional caching. Java table -writers generate sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits +writers and scans use sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits `manifest-sort.enabled`. Both ordinary writes and raw manifest rewrites build the sidecar from 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 do not yet invoke sidecar pruning automatically. Callers remain responsible for applying -entry filters and reconciling ADD/DELETE entries after block selection. The low-level `build` -method returns sidecar bytes without writing or publishing another file. +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. Selected block bytes share the manifest cache without populating the +whole-manifest entry cache with partial results. The low-level `build` method returns sidecar +bytes without writing or publishing another file. Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches. `build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java b/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java index 536be456eef0..3e564a60f9e1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java @@ -78,6 +78,22 @@ public boolean test(BinaryRow partition, int bucket, int totalBucket) { || totalAwareBucketFilter.test(partition, bucket, totalBucket); } + /** Conservatively checks an indexed pair without inventing a partition for custom filters. */ + public boolean mayContain(int bucket, int totalBuckets) { + if (onlyReadRealBuckets && bucket < 0) { + return false; + } + if (specifiedBucket != null && bucket != specifiedBucket) { + return false; + } + if (bucketFilter != null && !bucketFilter.test(bucket)) { + return false; + } + return !(totalAwareBucketFilter instanceof ManifestBucketFilter) + || ((ManifestBucketFilter) totalAwareBucketFilter) + .mayContain(bucket, bucket, totalBuckets); + } + /** Conservatively tests whether a manifest's bucket metadata can contain a matching entry. */ public boolean mayContain(ManifestFileMeta manifest) { Integer minBucket = manifest.minBucket(); 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 30e0adcf1bf3..81b4601ea414 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 @@ -37,6 +37,7 @@ import org.apache.paimon.utils.Filter; import org.apache.paimon.utils.ObjectsFile; import org.apache.paimon.utils.PathFactory; +import org.apache.paimon.utils.RowRangeIndex; import org.apache.paimon.utils.SegmentsCache; import javax.annotation.Nullable; @@ -140,9 +141,33 @@ public List read( Filter readFilter, Filter readTFilter, Function convertor) { + return read( + fileName, + fileSize, + partitionFilter, + bucketFilter, + readFilter, + readTFilter, + convertor, + null); + } + + public List read( + String fileName, + @Nullable Long fileSize, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + Filter readFilter, + Filter readTFilter, + Function convertor, + @Nullable ManifestSidecar.Selection selected) { + if (selected != null && selected.blocks().isEmpty()) { + return java.util.Collections.emptyList(); + } try { Path path = pathFactory.toPath(fileName); - if (cache != null) { + // Sidecar selections use the block cache, even when every block is selected. + if (cache != null && selected == null) { ManifestEntryFilters filters = new ManifestEntryFilters( partitionFilter, bucketFilter, readFilter, readTFilter); @@ -155,7 +180,9 @@ public List read( path, ManifestEntry.MANIFEST_ROW_TYPE, partitionFilter, - bucketFilter); + bucketFilter, + selected, + cache == null ? null : cache.segmentsCache()); return readFromIterator(iterator, serializer, readFilter, readTFilter, convertor); } catch (IOException e) { throw new UncheckedIOException(e); @@ -209,8 +236,23 @@ private static CloseableIterator createManifestIterator( @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter) throws IOException { + return createManifestIterator( + fileIO, path, projectedType, partitionFilter, bucketFilter, null, null); + } + + private static CloseableIterator createManifestIterator( + FileIO fileIO, + Path path, + RowType projectedType, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + @Nullable ManifestSidecar.Selection selected, + @Nullable SegmentsCache cache) + throws IOException { try { - ManifestAvroReader reader = new ManifestAvroReader(fileIO.newInputStream(path)); + ManifestAvroReader reader = + new ManifestAvroReader( + ManifestSidecar.openManifest(fileIO, path, selected, cache)); return reader.read(projectedType, partitionFilter, bucketFilter); } catch (IOException e) { FileUtils.checkExists(fileIO, path); @@ -345,6 +387,32 @@ public Path toPath(String fileName) { }; } + @Nullable + public ManifestSidecar.Selection selectBlocks( + ManifestFileMeta manifest, @Nullable RowRangeIndex query) { + return selectBlocks(manifest, query, null, null); + } + + @Nullable + public ManifestSidecar.Selection selectBlocks( + ManifestFileMeta manifest, + @Nullable RowRangeIndex query, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter) { + return !options.manifestSidecarEnabled() + || (query == null && partitionFilter == null && bucketFilter == null) + ? null + : ManifestSidecar.read( + fileIO, + pathFactory.toPath(manifest.fileName()), + manifest, + query, + partitionFilter, + partitionType, + bucketFilter == null ? null : bucketFilter::mayContain, + cache == null ? null : cache.segmentsCache()); + } + /** Deletes an unreferenced manifest and its explicitly referenced extra files. */ public void delete(ManifestFileMeta manifest) { delete(manifest.fileName()); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java index 82923ef9d827..e40d27e789b8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java @@ -29,6 +29,7 @@ import org.apache.paimon.manifest.ManifestEntrySerializer; import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestSidecar; import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.manifest.SimpleFileEntry; import org.apache.paimon.operation.metrics.ScanMetrics; @@ -498,27 +499,35 @@ private List readManifest( @Nullable Filter additionalFilter, @Nullable Filter additionalTFilter) { + ManifestFile manifestFile = manifestFileFactory.create(); + BucketFilter bucketFilter = createBucketFilter(); + ManifestSidecar.Selection selected = + manifestFile.selectBlocks( + manifest, rowRangeIndex, manifestsReader.partitionFilter(), bucketFilter); + if (selected != null && selected.blocks().isEmpty()) { + return Collections.emptyList(); + } Filter entryRowFilter = createEntryRowFilter(); Function finalConverter = dropStats ? e -> converter.apply(dropStats(e)) : converter; List entries = - manifestFileFactory - .create() + manifestFile .withCacheMetrics( scanMetrics != null ? scanMetrics.getCacheMetrics() : null) .read( manifest.fileName(), manifest.fileSize(), manifestsReader.partitionFilter(), - createBucketFilter(), + bucketFilter, entryRowFilter.and(additionalFilter), entry -> (additionalTFilter == null || additionalTFilter.test(entry)) && (manifestEntryFilter == null || manifestEntryFilter.test(entry)) && filterByStats(entry), - finalConverter); + finalConverter, + selected); LOG.info("Read {} manifest entries from {}", entries.size(), manifest.fileName()); return entries; } diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java index 4530f698e291..59106c5a2a70 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java @@ -63,6 +63,12 @@ public void withCacheMetrics(@Nullable CacheMetrics cacheMetrics) { this.cacheMetrics = cacheMetrics; } + /** Shares the byte cache with consumers using distinct whole-file and block keys. */ + @SuppressWarnings("unchecked") + public SegmentsCache segmentsCache() { + return (SegmentsCache) (SegmentsCache) cache; + } + public List read(K key, @Nullable Long fileSize, Filters filters) throws IOException { return read(key, fileSize, filters, Function.identity()); } 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 0362871ef826..15790dadaf87 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 @@ -19,6 +19,7 @@ package org.apache.paimon.manifest; import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.FileFormat; @@ -28,13 +29,18 @@ import org.apache.paimon.fs.FileIOFinder; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.SeekableInputStreamWrapper; import org.apache.paimon.fs.local.LocalFileIO; 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.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.StatsTestUtils; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; @@ -44,6 +50,8 @@ import org.apache.paimon.utils.FailingFileIO; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.Filter; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RowRangeIndex; import org.apache.paimon.utils.SegmentsCache; import org.junit.jupiter.api.RepeatedTest; @@ -68,13 +76,18 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; import java.util.stream.Stream; import static org.apache.paimon.TestKeyValueGenerator.DEFAULT_PART_TYPE; +import static org.apache.paimon.manifest.ManifestIndexTestUtils.withExtraFiles; import static org.apache.paimon.stats.StatsTestUtils.convertWithoutSchemaEvolution; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; /** Tests for {@link ManifestFile}. */ public class ManifestFileTest { @@ -1405,21 +1418,564 @@ private static int indexOf(byte[] bytes, byte[] target, int from, int limit) { return -1; } + @Test + void testSidecarCacheUsesExplicitPathsAndSharesTheManifestCache() throws Exception { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO io = new RecordingFileIO(); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io, cache); + List entries = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + ManifestEntry entry = gen.next(); + entries.add( + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(i * 1000000L))); + } + ManifestFileMeta written = factory.create().write(entries).get(0); + String sidecarName = "cached-explicit" + ManifestSidecar.SUFFIX; + java.nio.file.Path sidecar = tempDir.resolve("manifest").resolve(sidecarName); + java.nio.file.Files.move( + tempDir.resolve("manifest").resolve(ManifestSidecar.fileName(written)), sidecar); + ManifestFileMeta meta = withExtraFiles(written, Collections.singletonList(sidecarName)); + Path sidecarPath = new Path(tempDir.toString(), "manifest/" + sidecarName); + + io.reset(); + assertThat( + factory.create() + .selectBlocks( + meta, + RowRangeIndex.create( + Collections.singletonList( + new Range(Long.MAX_VALUE, Long.MAX_VALUE)))) + .blocks()) + .isEmpty(); + assertThat(io.opened).containsExactly(sidecarPath); + assertThat(cache.getIfPresents(sidecarPath).totalMemorySize()) + .isEqualTo(java.nio.file.Files.size(sidecar)); + + io.reset(); + RowRangeIndex hit = RowRangeIndex.create(Collections.singletonList(new Range(0, 0))); + assertThat(factory.create().selectBlocks(meta, hit).blocks()).isNotEmpty(); + assertThat(io.opened).isEmpty(); + assertThat(factory.create().read(meta.fileName())) + .containsExactlyInAnyOrderElementsOf(entries); + assertThat(cache.estimatedSize()).isEqualTo(2); + + io.reset(); + assertThat(factory.create().selectBlocks(meta, hit).blocks()).isNotEmpty(); + assertThat(factory.create().read(meta.fileName())) + .containsExactlyInAnyOrderElementsOf(entries); + assertThat(io.opened).isEmpty(); + } + + @Test + void testReadsOnlySelectedBlocksAndPreservesPhysicalOrdinals() throws Exception { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO fileIO = new RecordingFileIO(); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(16), Long.MAX_VALUE); + ManifestFile.Factory factory = + createManifestFileFactory( + tempDir.toString(), Long.MAX_VALUE, options, fileIO, cache); + ManifestFile manifests = factory.create(); + List entries = new ArrayList<>(); + for (int i = 0; i < 4000; i++) { + ManifestEntry entry = gen.next(); + entries.add( + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(i * 1000000L))); + } + ManifestFileMeta meta = manifests.write(entries).get(0); + Path manifestPath = new Path(tempDir.toString(), "manifest/" + meta.fileName()); + RowRangeIndex query = + RowRangeIndex.create( + Arrays.asList( + new Range(1000000000L, 1000000000L), + new Range(3000000000L, 3000000000L))); + + ManifestSidecar.Selection selected = manifests.selectBlocks(meta, query); + assertThat(selected.blocks()).hasSize(2); + + fileIO.reset(); + List actual = readSelectedEntries(factory.create(), meta, selected); + List expected = new ArrayList<>(); + for (ManifestSidecar.Block block : selected.blocks()) { + expected.addAll( + entries.subList( + (int) block.firstRecord, + (int) (block.firstRecord + block.recordCount))); + } + assertThat(actual).containsExactlyElementsOf(expected); + assertThat(actual).contains(entries.get(1000), entries.get(3000)); + assertThat(fileIO.bytes.get()).isLessThan(meta.fileSize() / 4); + assertThat(fileIO.seeks) + .containsExactlyElementsOf( + selected.blocks().stream() + .map(block -> block.offset) + .collect(Collectors.toList())); + assertThat(fileIO.opened).containsExactly(manifestPath); + assertThat(cache.getIfPresents(manifestPath)).isNull(); + fileIO.reset(); + assertThat(readSelectedEntries(factory.create(), meta, selected)) + .containsExactlyElementsOf(expected); + assertThat(fileIO.opened).isEmpty(); + assertThat(fileIO.bytes.get()).isZero(); + ManifestSidecar.Selection allBlocks = + manifests.selectBlocks( + meta, + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + fileIO.reset(); + assertThat(readSelectedEntries(factory.create(), meta, allBlocks)) + .containsExactlyInAnyOrderElementsOf(entries); + assertThat(fileIO.opened).containsExactly(manifestPath); + assertThat(cache.getIfPresents(manifestPath)).isNull(); + + fileIO.reset(); + assertThat(readSelectedEntries(factory.create(), meta, allBlocks)) + .containsExactlyInAnyOrderElementsOf(entries); + assertThat(fileIO.opened).isEmpty(); + assertThat(cache.getIfPresents(manifestPath)).isNull(); + + // Reads without a sidecar selection populate and reuse the full-manifest cache. + assertThat(manifests.read(meta.fileName())).containsExactlyInAnyOrderElementsOf(entries); + assertThat(fileIO.opened).containsExactly(manifestPath); + assertThat(cache.getIfPresents(manifestPath)).isNotNull(); + + fileIO.reset(); + assertThat(manifests.read(meta.fileName())).containsExactlyInAnyOrderElementsOf(entries); + assertThat(fileIO.opened).isEmpty(); + + long largestBlock = + allBlocks.blocks().stream().mapToLong(block -> block.length).max().getAsLong(); + assertThat(meta.fileSize()).isGreaterThan(largestBlock); + SegmentsCache blockCache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(16), largestBlock, null, false); + ManifestFile.Factory limitedFactory = + createManifestFileFactory( + tempDir.toString(), Long.MAX_VALUE, options, fileIO, blockCache); + for (int round = 0; round < 2; round++) { + fileIO.reset(); + assertThat(readSelectedEntries(limitedFactory.create(), meta, allBlocks)) + .containsExactlyElementsOf(entries); + if (round == 1) { + assertThat(fileIO.opened).isEmpty(); + } + } + assertThat( + blockCache.getIfPresents( + new Path(tempDir.toString(), "manifest/" + meta.fileName()))) + .isNull(); + } + + @Test + void testScannerPreservesDeletesAndColumnGroups() throws Exception { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO fileIO = new RecordingFileIO(); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, fileIO); + ManifestFile manifests = factory.create(); + ManifestEntry entry = gen.next(); + ManifestEntry add = + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(100L)); + ManifestEntry delete = + ManifestEntry.create( + FileKind.DELETE, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + add.file()); + ManifestEntry other = gen.next(); + ManifestEntry live = + ManifestEntry.create( + FileKind.ADD, + other.partition(), + other.bucket(), + other.totalBuckets(), + other.file().newFirstRowId(100L)); + List metas = new ArrayList<>(); + metas.addAll(manifests.write(Arrays.asList(add, live))); + metas.addAll(manifests.write(Collections.singletonList(delete))); + metas.addAll( + manifests.write( + Collections.singletonList( + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(0L))))); + AppendOnlyFileStoreScan scan = + new AppendOnlyFileStoreScan( + mock(ManifestsReader.class), + null, + null, + null, + mock(TableSchema.class), + factory, + 2, + false, + false, + false); + scan.withRowRanges(Collections.singletonList(new Range(100, 100))); + fileIO.reset(); + List result = new ArrayList<>(); + scan.readManifestEntries(metas, false).forEachRemaining(result::add); + assertThat(result).containsExactly(live); + assertThat( + fileIO.opened.stream() + .filter(path -> !path.getName().endsWith(ManifestSidecar.SUFFIX)) + .map(Path::getName)) + .containsExactlyInAnyOrder(metas.get(0).fileName(), metas.get(1).fileName()); + + for (boolean missing : new boolean[] {false, true}) { + for (ManifestFileMeta manifest : metas) { + Path sidecar = + new Path( + tempDir.toString(), + "manifest/" + ManifestSidecar.fileName(manifest)); + if (missing) { + fileIO.delete(sidecar, false); + } else { + fileIO.overwriteFileUtf8(sidecar, "corrupt sidecar"); + } + } + List fallback = new ArrayList<>(); + scan.readManifestEntries(metas, false).forEachRemaining(fallback::add); + assertThat(fallback).containsExactly(live); + } + } + + @Test + void testBucketOnlyPlanningAndRawRewriteUseNullableBucketPayload() throws Exception { + Options options = new Options(); + options.set(CoreOptions.BUCKET, 4); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO io = new RecordingFileIO(); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io); + ManifestFile manifests = factory.create(); + 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 = manifests.write(entries).get(0); + AppendOnlyFileStoreScan scan = + new AppendOnlyFileStoreScan( + mock(ManifestsReader.class), + null, + null, + null, + mock(TableSchema.class), + factory, + 2, + false, + false, + false); + scan.withBucket(1); + io.reset(); + assertThat(scan.readManifest(meta)).containsExactlyElementsOf(entries.subList(1000, 2000)); + assertThat(io.bytes.get()).isLessThan(meta.fileSize()); + assertThat(io.seeks).isNotEmpty(); + ManifestAvroWriter writer = manifests.createAvroWriter(); + try (ManifestAvroReader reader = + manifests.scanAvroBlocks(meta.fileName(), meta.fileSize())) { + writer.writeEncodedManifest(reader, meta); + } + writer.close(); + assertThat(scan.readManifest(writer.result().get(0))) + .containsExactlyElementsOf(entries.subList(1000, 2000)); + + ManifestEntry added = entries.get(1000); + ManifestEntry deleted = + ManifestEntry.create(FileKind.DELETE, added.partition(), 1, 4, added.file()); + List changes = new ArrayList<>(); + changes.addAll(manifests.write(Collections.singletonList(added))); + changes.addAll(manifests.write(Collections.singletonList(deleted))); + assertThat(scan.readManifestEntries(changes, false)).isExhausted(); + } + + @Test + void testPartitionOnlyPlanningUsesBlocksWithoutRowIds() { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO io = new RecordingFileIO(); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io); + BinaryRow first = gen.next().partition(); + BinaryRow second = gen.next().partition(); + while (second.equals(first)) { + second = gen.next().partition(); + } + List entries = new ArrayList<>(); + for (int i = 0; i < 4000; i++) { + ManifestEntry entry = gen.next(); + entries.add( + ManifestEntry.create( + FileKind.ADD, + i < 1000 ? first : second, + entry.bucket(), + entry.totalBuckets(), + entry.file())); + } + ManifestFileMeta meta = factory.create().write(entries).get(0); + ManifestsReader lists = mock(ManifestsReader.class); + when(lists.partitionFilter()) + .thenReturn( + PartitionPredicate.fromMultiple( + DEFAULT_PART_TYPE, Collections.singletonList(first))); + AppendOnlyFileStoreScan scan = + new AppendOnlyFileStoreScan( + lists, + null, + null, + null, + mock(TableSchema.class), + factory, + 2, + false, + false, + false); + io.reset(); + List actual = scan.readManifest(meta); + assertThat(actual).containsExactlyElementsOf(entries.subList(0, 1000)); + assertThat(io.bytes.get()).isLessThan(meta.fileSize()); + assertThat(io.opened).hasSize(2); + } + + @Test + void testUnknownRowIdKeepsPartitionIndexAndNoQueryDoesNotReadSidecar() { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO fileIO = new RecordingFileIO(); + ManifestFile manifests = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, fileIO) + .create(); + ManifestFileMeta meta = manifests.write(Collections.singletonList(gen.next())).get(0); + assertThat(ManifestSidecar.fileName(meta)).isNotNull(); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest") + .resolve(meta.fileName() + ManifestSidecar.SUFFIX))) + .isTrue(); + + fileIO.reset(); + assertThat(manifests.selectBlocks(meta, null)).isNull(); + assertThat(fileIO.opened).isEmpty(); + } + + @Test + void testExplicitIndexReferenceAndNullDoesNotProbe() throws Exception { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO io = new RecordingFileIO(); + ManifestFile manifests = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io).create(); + ManifestEntry original = gen.next(); + ManifestEntry entry = + ManifestEntry.create( + FileKind.ADD, + original.partition(), + original.bucket(), + original.totalBuckets(), + original.file().newFirstRowId(100L)); + ManifestFileMeta written = manifests.write(Collections.singletonList(entry)).get(0); + assertThat(ManifestSidecar.fileName(written)).isNotNull(); + RowRangeIndex query = RowRangeIndex.create(Collections.singletonList(new Range(0, 0))); + io.reset(); + for (List extraFiles : + Arrays.asList( + null, + Collections.emptyList(), + Collections.singletonList("other-partition-index"))) { + ManifestFileMeta unindexed = withExtraFiles(written, extraFiles); + assertThat(manifests.selectBlocks(unindexed, query)).isNull(); + assertThat(io.opened).isEmpty(); + } + // An existing suffix-named object must not be inferred as a reference. + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest") + .resolve(ManifestSidecar.fileName(written)))) + .isTrue(); + String explicitName = "custom-index-name" + ManifestSidecar.SUFFIX; + java.nio.file.Files.move( + tempDir.resolve("manifest").resolve(ManifestSidecar.fileName(written)), + tempDir.resolve("manifest").resolve(explicitName)); + String otherName = "other-partition-index"; + java.nio.file.Path otherPath = tempDir.resolve("manifest").resolve(otherName); + java.nio.file.Files.write(otherPath, new byte[] {1, 2, 3}); + ManifestFileMeta indexed = withExtraFiles(written, Arrays.asList(otherName, explicitName)); + assertThat(manifests.selectBlocks(indexed, query).blocks()).isEmpty(); + assertThat(io.opened) + .containsExactly(new Path(tempDir.toString(), "manifest/" + explicitName)); + manifests.delete(indexed); + assertThat(java.nio.file.Files.exists(otherPath)).isFalse(); + assertThat(java.nio.file.Files.exists(tempDir.resolve("manifest").resolve(explicitName))) + .isFalse(); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest").resolve(written.fileName()))) + .isFalse(); + } + + @Test + void testDisabledOrUnfilteredReadsSkipSidecarMetadata() { + for (boolean enabled : new boolean[] {false, true}) { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_SORT_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, enabled); + RecordingFileIO io = new RecordingFileIO(); + ManifestFile manifests = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io) + .create(); + ManifestFileMeta meta = mock(ManifestFileMeta.class); + RowRangeIndex rows = + enabled + ? null + : RowRangeIndex.create(Collections.singletonList(new Range(1, 1))); + assertThat(manifests.selectBlocks(meta, rows)).isNull(); + verifyNoInteractions(meta); + assertThat(io.opened).isEmpty(); + } + } + + private List readSelectedEntries( + ManifestFile manifests, ManifestFileMeta meta, ManifestSidecar.Selection selected) { + return manifests.read( + meta.fileName(), + meta.fileSize(), + null, + null, + row -> true, + entry -> true, + java.util.function.Function.identity(), + selected); + } + + /** Observes actual file access without adding counters to production readers. */ + private static final class RecordingFileIO extends LocalFileIO { + + private final List opened = Collections.synchronizedList(new ArrayList<>()); + private final List seeks = Collections.synchronizedList(new ArrayList<>()); + private final AtomicLong bytes = new AtomicLong(); + + private void reset() { + opened.clear(); + seeks.clear(); + bytes.set(0); + } + + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + opened.add(path); + return new SeekableInputStreamWrapper(super.newInputStream(path)) { + @Override + public void seek(long desired) throws IOException { + seeks.add(desired); + super.seek(desired); + } + + @Override + public int read() throws IOException { + int value = super.read(); + if (value >= 0) { + bytes.incrementAndGet(); + } + return value; + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + int n = super.read(buffer, offset, length); + if (n > 0) { + bytes.addAndGet(n); + } + return n; + } + }; + } + } + private ManifestFile createManifestFile(String pathStr) { return createManifestFile(pathStr, ThreadLocalRandom.current().nextInt(8192) + 1024); } private ManifestFile createManifestFile(String pathStr, long suggestedFileSize) { - return createManifestFile(pathStr, suggestedFileSize, null); + return createManifestFile(pathStr, suggestedFileSize, new Options()); } private ManifestFile createManifestFile( String pathStr, long suggestedFileSize, @Nullable SegmentsCache cache) { + return createManifestFileFactory( + pathStr, + suggestedFileSize, + new Options(), + FileIOFinder.find(new Path(pathStr)), + cache) + .create(); + } + + private ManifestFile createManifestFile( + String pathStr, long suggestedFileSize, Options options) { + return createManifestFileFactory( + pathStr, suggestedFileSize, options, FileIOFinder.find(new Path(pathStr))) + .create(); + } + + private ManifestFile.Factory createManifestFileFactory( + String pathStr, long suggestedFileSize, Options options, FileIO fileIO) { + return createManifestFileFactory(pathStr, suggestedFileSize, options, fileIO, null); + } + + private ManifestFile.Factory createManifestFileFactory( + String pathStr, + long suggestedFileSize, + Options options, + FileIO fileIO, + @Nullable SegmentsCache cache) { + return createManifestFileFactory( + pathStr, suggestedFileSize, options, fileIO, cache, DEFAULT_PART_TYPE); + } + + private ManifestFile.Factory createManifestFileFactory( + String pathStr, + long suggestedFileSize, + Options options, + FileIO fileIO, + @Nullable SegmentsCache cache, + RowType partitionType) { Path path = new Path(pathStr); FileStorePathFactory pathFactory = new FileStorePathFactory( path, - DEFAULT_PART_TYPE, + partitionType, "default", CoreOptions.FILE_FORMAT.defaultValue().toString(), CoreOptions.DATA_FILE_PREFIX.defaultValue(), @@ -1433,18 +1989,17 @@ private ManifestFile createManifestFile( null, false, null); - FileIO fileIO = FileIOFinder.find(path); + CoreOptions coreOptions = new CoreOptions(options); return new ManifestFile.Factory( - fileIO, - new FileSystemSchemaManager(fileIO, path), - DEFAULT_PART_TYPE, - avro, - "zstd", - pathFactory, - suggestedFileSize, - cache, - new CoreOptions(new Options())) - .create(); + fileIO, + new FileSystemSchemaManager(fileIO, path), + partitionType, + avro, + "zstd", + pathFactory, + suggestedFileSize, + cache, + coreOptions); } @Test From 640fd1b02614af77cdd4e1a72761b41aab941ae9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 16 Sep 2026 21:58:56 +0800 Subject: [PATCH 2/4] [core] Configure an independent manifest sidecar cache --- docs/docs/concepts/spec/manifest.md | 15 +- docs/generated/catalog_configuration.html | 6 + .../apache/paimon/options/CatalogOptions.java | 10 ++ .../ManifestFileSorterBenchmark.java | 1 + .../org/apache/paimon/AbstractFileStore.java | 7 + .../java/org/apache/paimon/FileStore.java | 2 + .../apache/paimon/catalog/CachingCatalog.java | 21 +++ .../apache/paimon/manifest/ManifestFile.java | 13 +- .../paimon/manifest/ManifestSidecar.java | 4 +- .../paimon/table/AbstractFileStoreTable.java | 16 ++ .../paimon/table/DelegatedFileStoreTable.java | 11 ++ .../table/FallbackReadFileStoreTable.java | 6 + .../apache/paimon/table/FileStoreTable.java | 5 + .../paimon/catalog/CachingCatalogTest.java | 138 ++++++++++++++++++ .../paimon/manifest/ManifestFileMetaTest.java | 2 + .../manifest/ManifestFileMetaTestBase.java | 1 + .../paimon/manifest/ManifestFileTest.java | 106 ++++++++++++-- .../paimon/manifest/ManifestSidecarTest.java | 12 +- .../manifest/ManifestSidecarWriteTest.java | 1 + .../operation/ManifestRewriteCleanupTest.java | 1 + 20 files changed, 351 insertions(+), 27 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index d05a90e9a477..0bfbaaa4bacc 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -79,16 +79,21 @@ close successfully. Failed writes and aborted writers clean up their own manifes 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. Selected block bytes share the manifest cache without populating the -whole-manifest entry cache with partial results. The low-level `build` method returns sidecar -bytes without writing or publishing another file. +perform sidecar I/O. Sidecar caching is controlled by the catalog option +`cache.manifest-sidecar.max-memory` (0 by default). A positive value supplies an +additional budget independent of the manifest content cache; 0 reuses the manifest content +cache, or leaves sidecars uncached if that cache is disabled. Sidecar caching uses the catalog's +`cache.expire-after-access` and `cache.manifest.soft-values` policies. Selected block bytes +still share the manifest content cache without populating the whole-manifest entry cache +with partial results. The low-level `build` method returns sidecar bytes without writing +or publishing another file. Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches. `build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent payload generation. Partition generation is always enabled, including the empty partition tuple for unpartitioned tables. Missing or invalid -metadata makes only the affected block's dimension unavailable. There is no sidecar byte budget: -construction keeps complete coverage and `read` consumes the entire file once it is opened. +metadata makes only the affected block's dimension unavailable. Cache limits do not truncate +sidecars: construction keeps complete coverage and `read` consumes the entire file once it is opened. `read` returns null for an absent sidecar reference or an `IOException`, allowing the caller to fall back to the manifest. If the thread is interrupted, the I/O failure is propagated as diff --git a/docs/generated/catalog_configuration.html b/docs/generated/catalog_configuration.html index 806edac3f870..e056586b339d 100644 --- a/docs/generated/catalog_configuration.html +++ b/docs/generated/catalog_configuration.html @@ -50,6 +50,12 @@ Duration Cache expiration policy: marks cache entries to expire after a specified duration has passed since their last refresh. + +
cache.manifest-sidecar.max-memory
+ 0 bytes + MemorySize + Controls the maximum memory for the separate manifest sidecar cache. This budget is additional to the manifest content cache. Set to 0 to reuse the manifest content cache. It uses 'cache.expire-after-access' and 'cache.manifest.soft-values'. +
cache.manifest.max-memory
(none) diff --git a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java index b1ccfe1caada..e096617b0da3 100644 --- a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java @@ -130,6 +130,16 @@ public class CatalogOptions { .noDefaultValue() .withDescription("Controls the maximum memory to cache manifest content."); + public static final ConfigOption CACHE_MANIFEST_SIDECAR_MAX_MEMORY = + key("cache.manifest-sidecar.max-memory") + .memoryType() + .defaultValue(MemorySize.ofBytes(0)) + .withDescription( + "Controls the maximum memory for the separate manifest sidecar cache. " + + "This budget is additional to the manifest content cache. " + + "Set to 0 to reuse the manifest content cache. It uses " + + "'cache.expire-after-access' and 'cache.manifest.soft-values'."); + public static final ConfigOption CACHE_MANIFEST_SOFT_VALUES = key("cache.manifest.soft-values") .booleanType() diff --git a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java index e805efb69443..1f78ff8d4591 100644 --- a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java +++ b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java @@ -301,6 +301,7 @@ private ManifestFile createManifestFile() { pathFactory, TARGET_MANIFEST_SIZE, null, + null, new CoreOptions(new Options())) .create(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index 29c7ab8de037..86839355a718 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -109,6 +109,7 @@ abstract class AbstractFileStore implements FileStore { protected final CatalogEnvironment catalogEnvironment; @Nullable private SegmentsCache readManifestCache; + @Nullable private SegmentsCache manifestSidecarCache; @Nullable private Cache snapshotCache; protected AbstractFileStore( @@ -212,6 +213,7 @@ public ManifestFile.Factory manifestFileFactory() { pathFactory(), options.manifestTargetSize().getBytes(), readManifestCache, + manifestSidecarCache, options); } @@ -616,6 +618,11 @@ public void setManifestCache(SegmentsCache manifestCache) { this.readManifestCache = manifestCache; } + @Override + public void setManifestSidecarCache(SegmentsCache manifestSidecarCache) { + this.manifestSidecarCache = manifestSidecarCache; + } + @Override public void setSnapshotCache(Cache cache) { this.snapshotCache = cache; diff --git a/paimon-core/src/main/java/org/apache/paimon/FileStore.java b/paimon-core/src/main/java/org/apache/paimon/FileStore.java index 1714eec2a983..d9b143f75f59 100644 --- a/paimon-core/src/main/java/org/apache/paimon/FileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/FileStore.java @@ -127,5 +127,7 @@ boolean mergeSchema( void setManifestCache(SegmentsCache manifestCache); + void setManifestSidecarCache(SegmentsCache manifestSidecarCache); + void setSnapshotCache(Cache cache); } diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java index 7fce5edf0ef4..037f7ea2bfba 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java @@ -51,6 +51,7 @@ import static org.apache.paimon.options.CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS; import static org.apache.paimon.options.CatalogOptions.CACHE_EXPIRE_AFTER_WRITE; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_MAX_MEMORY; +import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SIDECAR_MAX_MEMORY; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_MEMORY; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_THRESHOLD; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SOFT_VALUES; @@ -71,6 +72,7 @@ public class CachingCatalog extends DelegateCatalog { protected Cache databaseCache; protected Cache tableCache; @Nullable protected final SegmentsCache manifestCache; + @Nullable protected final SegmentsCache manifestSidecarCache; // partition cache will affect data latency @Nullable protected Cache> partitionCache; @Nullable protected DVMetaCache dvMetaCache; @@ -108,6 +110,17 @@ public CachingCatalog(Catalog wrapped, Options options) { expireAfterAccess, manifestCacheSoftValues); + MemorySize sidecarMaxMemory = options.get(CACHE_MANIFEST_SIDECAR_MAX_MEMORY); + this.manifestSidecarCache = + sidecarMaxMemory.getBytes() == 0 + ? manifestCache + : SegmentsCache.create( + (int) CoreOptions.PAGE_SIZE.defaultValue().getBytes(), + sidecarMaxMemory, + sidecarMaxMemory.getBytes(), + expireAfterAccess, + manifestCacheSoftValues); + this.cachedPartitionMaxNum = options.get(CACHE_PARTITION_MAX_NUM); int cacheDvMaxNum = options.get(CACHE_DV_MAX_NUM); @@ -299,6 +312,9 @@ private Table loadTable(Identifier identifier) { if (manifestCache != null) { storeTable.setManifestCache(manifestCache); } + if (manifestSidecarCache != null) { + storeTable.setManifestSidecarCache(manifestSidecarCache); + } if (dvMetaCache != null) { storeTable.setDVMetaCache(dvMetaCache); } @@ -425,6 +441,11 @@ public CacheSizes estimatedCacheSizes() { manifestCacheSize = manifestCache.estimatedSize(); manifestCacheBytes = manifestCache.totalCacheBytes(); } + // Keep reporting total manifest metadata usage even though sidecars have their own budget. + if (manifestSidecarCache != null && manifestSidecarCache != manifestCache) { + manifestCacheSize += manifestSidecarCache.estimatedSize(); + manifestCacheBytes += manifestSidecarCache.totalCacheBytes(); + } long partitionCacheSize = 0L; if (partitionCache != null) { for (Map.Entry> entry : partitionCache.asMap().entrySet()) { 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 81b4601ea414..9740961e73d2 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 @@ -62,6 +62,7 @@ public class ManifestFile extends ObjectsFile { private final AvroFileFormat avroFileFormat; private final long suggestedFileSize; private final CoreOptions options; + @Nullable private final SegmentsCache sidecarCache; private ManifestFile( FileIO fileIO, @@ -73,6 +74,7 @@ private ManifestFile( PathFactory pathFactory, long suggestedFileSize, @Nullable SegmentsCache cache, + @Nullable SegmentsCache sidecarCache, CoreOptions options) { super( fileIO, @@ -90,6 +92,7 @@ private ManifestFile( this.avroFileFormat = avroFileFormat; this.suggestedFileSize = suggestedFileSize; this.options = options; + this.sidecarCache = sidecarCache == null ? cache : sidecarCache; } @Override @@ -410,7 +413,7 @@ public ManifestSidecar.Selection selectBlocks( partitionFilter, partitionType, bucketFilter == null ? null : bucketFilter::mayContain, - cache == null ? null : cache.segmentsCache()); + sidecarCache); } /** Deletes an unreferenced manifest and its explicitly referenced extra files. */ @@ -433,6 +436,7 @@ public static class Factory { private final long suggestedFileSize; private final CoreOptions options; @Nullable private final SegmentsCache cache; + @Nullable private final SegmentsCache sidecarCache; public Factory( FileIO fileIO, @@ -443,6 +447,7 @@ public Factory( FileStorePathFactory pathFactory, long suggestedFileSize, @Nullable SegmentsCache cache, + @Nullable SegmentsCache sidecarCache, CoreOptions options) { this.fileIO = fileIO; this.schemaManager = schemaManager; @@ -452,13 +457,10 @@ public Factory( this.pathFactory = pathFactory; this.suggestedFileSize = suggestedFileSize; this.cache = cache; + this.sidecarCache = sidecarCache; this.options = options; } - public boolean isCacheEnabled() { - return cache != null; - } - public ManifestFile create() { return new ManifestFile( fileIO, @@ -470,6 +472,7 @@ public ManifestFile create() { pathFactory.manifestFileFactory(), suggestedFileSize, cache, + sidecarCache, options); } } 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 c6ee9e9f5615..6a8ce2b143bb 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 @@ -632,7 +632,7 @@ public static Selection read( @Nullable PartitionPredicate partitionFilter, @Nullable RowType partitionType, @Nullable BiPredicate bucketFilter, - @Nullable SegmentsCache cache) { + @Nullable SegmentsCache cache) { String sidecarFileName = fileName(manifest); if (sidecarFileName == null) { return null; @@ -660,7 +660,7 @@ public static Selection read( } } - /** Complete sidecar bytes stored in the shared manifest cache. */ + /** Complete sidecar bytes stored in the configured sidecar cache. */ static final class ManifestSidecarSegment implements Segments { private final byte[] bytes; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index e9d9603f5a98..93ecc173d5d2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -107,6 +107,7 @@ abstract class AbstractFileStoreTable implements FileStoreTable { @Nullable private Set appliedDynamicOptionKeys; @Nullable protected transient SegmentsCache manifestCache; + @Nullable protected transient SegmentsCache manifestSidecarCache; @Nullable protected transient Cache snapshotCache; @Nullable protected transient Cache statsCache; @Nullable protected transient DVMetaCache dvmetaCache; @@ -144,6 +145,18 @@ public SegmentsCache getManifestCache() { return manifestCache; } + @Override + public void setManifestSidecarCache(SegmentsCache manifestSidecarCache) { + this.manifestSidecarCache = manifestSidecarCache; + store().setManifestSidecarCache(manifestSidecarCache); + } + + @Nullable + @Override + public SegmentsCache getManifestSidecarCache() { + return manifestSidecarCache; + } + @Override public void setSnapshotCache(Cache cache) { this.snapshotCache = cache; @@ -432,6 +445,9 @@ public FileStoreTable copy(TableSchema newTableSchema) { if (manifestCache != null) { copied.setManifestCache(manifestCache); } + if (manifestSidecarCache != null) { + copied.setManifestSidecarCache(manifestSidecarCache); + } if (statsCache != null) { copied.setStatsCache(statsCache); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java index 9a5e81bcb636..ed541721fca0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java @@ -145,6 +145,17 @@ public SegmentsCache getManifestCache() { return wrapped.getManifestCache(); } + @Override + public void setManifestSidecarCache(SegmentsCache manifestSidecarCache) { + wrapped.setManifestSidecarCache(manifestSidecarCache); + } + + @Nullable + @Override + public SegmentsCache getManifestSidecarCache() { + return wrapped.getManifestSidecarCache(); + } + @Override public void setSnapshotCache(Cache cache) { wrapped.setSnapshotCache(cache); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java index da37e0780825..940dd38b2b63 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java @@ -149,6 +149,12 @@ public void setManifestCache(SegmentsCache manifestCache) { other.setManifestCache(manifestCache); } + @Override + public void setManifestSidecarCache(SegmentsCache manifestSidecarCache) { + super.setManifestSidecarCache(manifestSidecarCache); + other.setManifestSidecarCache(manifestSidecarCache); + } + protected FileStoreTable switchWrappedToBranch(String branchName) { Optional optionalSchema = wrapped.schemaManager().copyWithBranch(branchName).latest(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java index 970493430589..ddaa17407624 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java @@ -70,6 +70,11 @@ interface SnapshotReaderFactory { @Nullable SegmentsCache getManifestCache(); + void setManifestSidecarCache(SegmentsCache manifestSidecarCache); + + @Nullable + SegmentsCache getManifestSidecarCache(); + void setSnapshotCache(Cache cache); void setStatsCache(Cache cache); diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index 3de7bbf73f16..358ef449f8f0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -22,12 +22,17 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.data.GenericRow; import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestSidecar; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.partition.PartitionStatistics; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.sink.BatchTableWrite; @@ -48,6 +53,8 @@ import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.Mockito; import java.io.FileNotFoundException; @@ -71,6 +78,7 @@ import static org.apache.paimon.data.BinaryString.fromString; import static org.apache.paimon.options.CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_MAX_MEMORY; +import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SIDECAR_MAX_MEMORY; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_MEMORY; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_THRESHOLD; import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SOFT_VALUES; @@ -633,4 +641,134 @@ public void testManifestCacheOptions() { assertThat(caching.manifestCache.softValues()).isFalse(); assertThat(caching.manifestCache.ttl()).isEqualTo(CACHE_EXPIRE_AFTER_ACCESS.defaultValue()); } + + @Test + public void testManifestSidecarCacheOptions() { + Options options = new Options(); + CachingCatalog caching = new CachingCatalog(catalog, options); + assertThat(CACHE_MANIFEST_SIDECAR_MAX_MEMORY.defaultValue()) + .isEqualTo(MemorySize.ofBytes(0)); + assertThat(caching.manifestSidecarCache).isSameAs(caching.manifestCache); + + options.set(CACHE_MANIFEST_SIDECAR_MAX_MEMORY, MemorySize.ofMebiBytes(8)); + caching = new CachingCatalog(catalog, options); + assertThat(caching.manifestSidecarCache).isNotSameAs(caching.manifestCache); + assertThat(caching.manifestSidecarCache.maxMemorySize()) + .isEqualTo(MemorySize.ofMebiBytes(8)); + assertThat(caching.manifestSidecarCache.maxElementSize()) + .isEqualTo(MemorySize.ofMebiBytes(8).getBytes()); + assertThat(caching.manifestSidecarCache.ttl()) + .isEqualTo(CACHE_EXPIRE_AFTER_ACCESS.defaultValue()); + assertThat(caching.manifestSidecarCache.softValues()).isTrue(); + + options.set(CACHE_MANIFEST_SMALL_FILE_MEMORY, MemorySize.ofBytes(0)); + options.set(CACHE_EXPIRE_AFTER_ACCESS, Duration.ofMinutes(2)); + options.set(CACHE_MANIFEST_SOFT_VALUES, false); + caching = new CachingCatalog(catalog, options); + assertThat(caching.manifestCache).isNull(); + assertThat(caching.manifestSidecarCache.maxMemorySize()) + .isEqualTo(MemorySize.ofMebiBytes(8)); + assertThat(caching.manifestSidecarCache.ttl()).isEqualTo(Duration.ofMinutes(2)); + assertThat(caching.manifestSidecarCache.softValues()).isFalse(); + + options.set(CACHE_MANIFEST_SIDECAR_MAX_MEMORY, MemorySize.ofBytes(0)); + options.set(CACHE_MANIFEST_SMALL_FILE_MEMORY, MemorySize.ofMebiBytes(1)); + caching = new CachingCatalog(catalog, options); + assertThat(caching.manifestCache).isNotNull(); + assertThat(caching.manifestSidecarCache).isSameAs(caching.manifestCache); + + options.set(CACHE_MANIFEST_SMALL_FILE_MEMORY, MemorySize.ofBytes(0)); + caching = new CachingCatalog(catalog, options); + assertThat(caching.manifestCache).isNull(); + assertThat(caching.manifestSidecarCache).isNull(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testManifestSidecarCacheSharedAcrossTableCopies(boolean separate) throws Exception { + Options options = new Options(); + options.set(CACHE_MANIFEST_SOFT_VALUES, false); + options.set( + CACHE_MANIFEST_SIDECAR_MAX_MEMORY, + separate ? MemorySize.ofMebiBytes(1) : MemorySize.ofBytes(0)); + CachingCatalog caching = new CachingCatalog(catalog, options); + Identifier id = Identifier.create("db", "sidecar_cache"); + caching.createTable(id, DEFAULT_TABLE_SCHEMA, false); + caching.alterTable( + id, + SchemaChange.setOption(CoreOptions.MANIFEST_SIDECAR_ENABLED.key(), "true"), + false); + FileStoreTable table = (FileStoreTable) caching.getTable(id); + assertThat(table.getManifestSidecarCache()).isSameAs(caching.manifestSidecarCache); + writeTableForTestManifestCache(table); + ManifestFileMeta meta = + table.store() + .manifestListFactory() + .create() + .readDataManifests(table.latestSnapshot().get()) + .get(0); + PartitionPredicate partition = PartitionPredicate.ALWAYS_TRUE; + assertThat( + table.store() + .manifestFileFactory() + .create() + .selectBlocks(meta, null, partition, null)) + .isNotNull(); + Path sidecar = new Path(table.location(), "manifest/" + ManifestSidecar.fileName(meta)); + assertThat(caching.manifestSidecarCache.getIfPresents(sidecar)).isNotNull(); + if (separate) { + assertThat(caching.manifestCache.getIfPresents(sidecar)).isNull(); + } else { + assertThat(caching.manifestCache.getIfPresents(sidecar)).isNotNull(); + } + assertThat(caching.estimatedCacheSizes().manifestCacheSize()) + .isEqualTo( + caching.manifestCache.estimatedSize() + + (separate ? caching.manifestSidecarCache.estimatedSize() : 0)); + assertThat(caching.estimatedCacheSizes().manifestCacheBytes()) + .isEqualTo( + caching.manifestCache.totalCacheBytes() + + (separate ? caching.manifestSidecarCache.totalCacheBytes() : 0)); + + // A copied table and a new ManifestFile must reuse cached bytes even after the file is + // gone. + fileIO.delete(sidecar, false); + for (FileStoreTable copy : + new FileStoreTable[] { + table.copy(singletonMap("a", "b")), + table.copy(table.schema()), + (FileStoreTable) caching.getTable(id) + }) { + assertThat(copy.getManifestSidecarCache()).isSameAs(caching.manifestSidecarCache); + assertThat( + copy.store() + .manifestFileFactory() + .create() + .selectBlocks(meta, null, partition, null)) + .isNotNull(); + } + } + + @Test + public void testManifestSidecarCachePropagatesToFallbackTable() throws Exception { + CachingCatalog caching = new CachingCatalog(catalog, new Options()); + Identifier id = Identifier.create("db", "sidecar_fallback"); + caching.createTable(id, DEFAULT_TABLE_SCHEMA, false); + caching.getTable(id).createBranch("fallback"); + caching.alterTable( + id, + SchemaChange.setOption(CoreOptions.SCAN_FALLBACK_BRANCH.key(), "fallback"), + false); + FallbackReadFileStoreTable table = (FallbackReadFileStoreTable) caching.getTable(id); + for (FallbackReadFileStoreTable copy : + new FallbackReadFileStoreTable[] { + table, (FallbackReadFileStoreTable) table.copy(singletonMap("a", "b")) + }) { + assertThat(copy.getManifestSidecarCache()).isSameAs(caching.manifestSidecarCache); + assertThat(copy.wrapped().getManifestSidecarCache()) + .isSameAs(caching.manifestSidecarCache); + assertThat(copy.other().getManifestSidecarCache()) + .isSameAs(caching.manifestSidecarCache); + } + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java index 16f33f4cee97..161591a48e11 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java @@ -2745,6 +2745,7 @@ public void testManifestSortWithMultiplePartitions() { null), Long.MAX_VALUE, null, + null, new CoreOptions(new Options())) .create(); @@ -3206,6 +3207,7 @@ private ManifestFile createManifestFileForPartitionType(RowType partitionType) { null), Long.MAX_VALUE, null, + null, new CoreOptions(new Options())) .create(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java index a4516015c29b..4e72f58e7382 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java @@ -167,6 +167,7 @@ protected ManifestFile createManifestFile(String pathStr, FileIO fileIO) { null), Long.MAX_VALUE, null, + null, new CoreOptions(new Options())) .create(); } 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 15790dadaf87..db914239eea5 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 @@ -1419,15 +1419,18 @@ private static int indexOf(byte[] bytes, byte[] target, int from, int limit) { } @Test - void testSidecarCacheUsesExplicitPathsAndSharesTheManifestCache() throws Exception { + void testSidecarCacheUsesExplicitPathsAndIsSeparateFromManifestCache() throws Exception { Options options = new Options(); options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); RecordingFileIO io = new RecordingFileIO(); SegmentsCache cache = new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); + SegmentsCache sidecarCache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); ManifestFile.Factory factory = - createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io, cache); + createManifestFileFactory( + tempDir.toString(), Long.MAX_VALUE, options, io, cache, sidecarCache); List entries = new ArrayList<>(); for (int i = 0; i < 2; i++) { ManifestEntry entry = gen.next(); @@ -1458,8 +1461,9 @@ void testSidecarCacheUsesExplicitPathsAndSharesTheManifestCache() throws Excepti .blocks()) .isEmpty(); assertThat(io.opened).containsExactly(sidecarPath); - assertThat(cache.getIfPresents(sidecarPath).totalMemorySize()) + assertThat(sidecarCache.getIfPresents(sidecarPath).totalMemorySize()) .isEqualTo(java.nio.file.Files.size(sidecar)); + assertThat(cache.estimatedSize()).isZero(); io.reset(); RowRangeIndex hit = RowRangeIndex.create(Collections.singletonList(new Range(0, 0))); @@ -1467,7 +1471,13 @@ void testSidecarCacheUsesExplicitPathsAndSharesTheManifestCache() throws Excepti assertThat(io.opened).isEmpty(); assertThat(factory.create().read(meta.fileName())) .containsExactlyInAnyOrderElementsOf(entries); - assertThat(cache.estimatedSize()).isEqualTo(2); + assertThat(cache.estimatedSize()).isEqualTo(1); + assertThat(sidecarCache.estimatedSize()).isEqualTo(1); + assertThat(cache.getIfPresents(sidecarPath)).isNull(); + assertThat( + sidecarCache.getIfPresents( + new Path(tempDir.toString(), "manifest/" + meta.fileName()))) + .isNull(); io.reset(); assertThat(factory.create().selectBlocks(meta, hit).blocks()).isNotEmpty(); @@ -1476,6 +1486,84 @@ void testSidecarCacheUsesExplicitPathsAndSharesTheManifestCache() throws Excepti assertThat(io.opened).isEmpty(); } + @Test + void testDedicatedSidecarCacheAndManifestCacheFallback() { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + for (boolean cacheManifest : new boolean[] {false, true}) { + for (boolean cacheSidecar : new boolean[] {false, true}) { + RecordingFileIO io = new RecordingFileIO(); + SegmentsCache manifestCache = + cacheManifest + ? new SegmentsCache<>( + 1024, + MemorySize.ofMebiBytes(1), + Long.MAX_VALUE, + null, + false) + : null; + SegmentsCache sidecarCache = + cacheSidecar + ? new SegmentsCache<>( + 1024, + MemorySize.ofMebiBytes(1), + Long.MAX_VALUE, + null, + false) + : null; + ManifestFile.Factory factory = + createManifestFileFactory( + tempDir.resolve(cacheManifest + "-" + cacheSidecar).toString(), + Long.MAX_VALUE, + options, + io, + manifestCache, + sidecarCache); + ManifestEntry entry = gen.next(); + ManifestEntry added = + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(0L)); + ManifestFileMeta meta = + factory.create().write(Collections.singletonList(added)).get(0); + RowRangeIndex query = + RowRangeIndex.create(Collections.singletonList(new Range(0, 0))); + + for (int round = 0; round < 2; round++) { + io.reset(); + ManifestFile manifest = factory.create(); + ManifestSidecar.Selection selected = manifest.selectBlocks(meta, query); + assertThat(readSelectedEntries(manifest, meta, selected)) + .containsExactly(added); + assertThat( + io.opened.stream() + .filter( + path -> + path.getName() + .endsWith( + ManifestSidecar + .SUFFIX))) + .hasSize(round == 0 || (!cacheSidecar && !cacheManifest) ? 1 : 0); + assertThat( + io.opened.stream() + .filter(path -> path.getName().equals(meta.fileName()))) + .hasSize(round == 0 || !cacheManifest ? 1 : 0); + } + if (manifestCache != null) { + // Blocks stay in the manifest cache; sidecar bytes only join them on fallback. + assertThat(manifestCache.estimatedSize()).isEqualTo(cacheSidecar ? 1 : 2); + } + if (sidecarCache != null) { + assertThat(sidecarCache.estimatedSize()).isEqualTo(1); + } + } + } + } + @Test void testReadsOnlySelectedBlocksAndPreservesPhysicalOrdinals() throws Exception { Options options = new Options(); @@ -1960,8 +2048,7 @@ private ManifestFile.Factory createManifestFileFactory( Options options, FileIO fileIO, @Nullable SegmentsCache cache) { - return createManifestFileFactory( - pathStr, suggestedFileSize, options, fileIO, cache, DEFAULT_PART_TYPE); + return createManifestFileFactory(pathStr, suggestedFileSize, options, fileIO, cache, null); } private ManifestFile.Factory createManifestFileFactory( @@ -1970,12 +2057,12 @@ private ManifestFile.Factory createManifestFileFactory( Options options, FileIO fileIO, @Nullable SegmentsCache cache, - RowType partitionType) { + @Nullable SegmentsCache sidecarCache) { Path path = new Path(pathStr); FileStorePathFactory pathFactory = new FileStorePathFactory( path, - partitionType, + DEFAULT_PART_TYPE, "default", CoreOptions.FILE_FORMAT.defaultValue().toString(), CoreOptions.DATA_FILE_PREFIX.defaultValue(), @@ -1993,12 +2080,13 @@ private ManifestFile.Factory createManifestFileFactory( return new ManifestFile.Factory( fileIO, new FileSystemSchemaManager(fileIO, path), - partitionType, + DEFAULT_PART_TYPE, avro, "zstd", pathFactory, suggestedFileSize, cache, + sidecarCache, coreOptions); } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java index 1110c7015861..1b3f9d258ee4 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java @@ -458,14 +458,14 @@ void cacheRespectsElementThreshold() throws Exception { Files.write(temp.resolve(sidecar.getName()), data); ManifestFileMeta meta = testMeta(); FileIO io = spy(LocalFileIO.create()); - SegmentsCache tooSmall = + SegmentsCache tooSmall = new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), data.length - 1L, null, false); assertThat(readCached(io, path, meta, tooSmall).blocks()).hasSize(2); assertThat(readCached(io, path, meta, tooSmall).blocks()).hasSize(2); assertThat(tooSmall.getIfPresents(sidecar)).isNull(); verify(io, times(2)).newInputStream(sidecar); - SegmentsCache cache = + SegmentsCache cache = new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), data.length, null, false); assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); @@ -478,7 +478,7 @@ void cachedSegmentsRequireSidecarType() throws Exception { Path path = new Path(temp.toString(), "manifest-golden"); Path sidecar = ManifestSidecar.path(path); Files.write(temp.resolve(sidecar.getName()), data); - SegmentsCache cache = + SegmentsCache cache = new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); cache.put(sidecar, new SingleSegments(MemorySegment.wrap(data), data.length)); FileIO io = spy(LocalFileIO.create()); @@ -499,7 +499,7 @@ void missingAndInvalidSidecarsAreNotCached() throws Exception { Path sidecar = ManifestSidecar.path(path); ManifestFileMeta meta = testMeta(); FileIO io = spy(LocalFileIO.create()); - SegmentsCache cache = + SegmentsCache cache = new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); assertThat(readCached(io, path, meta, cache)).isNull(); assertThat(cache.getIfPresents(sidecar)).isNull(); @@ -521,7 +521,7 @@ void cachedBytesPreservePerQueryCancellation() throws Exception { Files.write(temp.resolve(sidecar.getName()), testSidecar()); ManifestFileMeta meta = testMeta(); FileIO io = spy(LocalFileIO.create()); - SegmentsCache cache = + SegmentsCache cache = new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); RowRangeIndex cancelled = mock(RowRangeIndex.class); when(cancelled.intersects(anyLong(), anyLong())) @@ -544,7 +544,7 @@ void cachedBytesPreservePerQueryCancellation() throws Exception { } private ManifestSidecar.Selection readCached( - FileIO io, Path path, ManifestFileMeta meta, SegmentsCache cache) { + FileIO io, Path path, ManifestFileMeta meta, SegmentsCache cache) { return ManifestSidecar.read( io, path, diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarWriteTest.java index cea952943f59..3549bf6dcda1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarWriteTest.java @@ -341,6 +341,7 @@ private ManifestFile manifests( paths, targetSize, null, + null, new CoreOptions(options)) .create(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java index 20a1b26beece..dd1a1b176b96 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java @@ -669,6 +669,7 @@ private ManifestFile createManifestFile(long suggestedFileSize) { null), suggestedFileSize, null, + null, new CoreOptions(new Options())) .create(); } From 994f824994bc3aecc725998cd78e6e57b11793b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 16 Sep 2026 22:10:15 +0800 Subject: [PATCH 3/4] [core] Leave manifest sidecar cache size unset by default --- docs/docs/concepts/spec/manifest.md | 8 ++++---- docs/generated/catalog_configuration.html | 4 ++-- .../java/org/apache/paimon/options/CatalogOptions.java | 4 ++-- .../java/org/apache/paimon/catalog/CachingCatalog.java | 4 +++- .../org/apache/paimon/catalog/CachingCatalogTest.java | 4 ++-- 5 files changed, 13 insertions(+), 11 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 0bfbaaa4bacc..ecea86a47097 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -80,10 +80,10 @@ Scans with partition, row-ID or bucket filters select blocks before reading mani 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 -`cache.manifest-sidecar.max-memory` (0 by default). A positive value supplies an -additional budget independent of the manifest content cache; 0 reuses the manifest content -cache, or leaves sidecars uncached if that cache is disabled. Sidecar caching uses the catalog's -`cache.expire-after-access` and `cache.manifest.soft-values` policies. Selected block bytes +`cache.manifest-sidecar.max-memory` (unset by default). A positive value supplies an +additional budget independent of the manifest content cache. When unset or set to 0, sidecars +reuse the manifest content cache, or remain uncached if that cache is disabled. Sidecar caching +uses the catalog's `cache.expire-after-access` and `cache.manifest.soft-values` policies. Selected block bytes still share the manifest content cache without populating the whole-manifest entry cache with partial results. The low-level `build` method returns sidecar bytes without writing or publishing another file. diff --git a/docs/generated/catalog_configuration.html b/docs/generated/catalog_configuration.html index e056586b339d..84c59561cdc6 100644 --- a/docs/generated/catalog_configuration.html +++ b/docs/generated/catalog_configuration.html @@ -52,9 +52,9 @@
cache.manifest-sidecar.max-memory
- 0 bytes + (none) MemorySize - Controls the maximum memory for the separate manifest sidecar cache. This budget is additional to the manifest content cache. Set to 0 to reuse the manifest content cache. It uses 'cache.expire-after-access' and 'cache.manifest.soft-values'. + Controls the maximum memory for the separate manifest sidecar cache. This budget is additional to the manifest content cache. When unset or set to 0, reuses the manifest content cache. It uses 'cache.expire-after-access' and 'cache.manifest.soft-values'.
cache.manifest.max-memory
diff --git a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java index e096617b0da3..09b3c934c584 100644 --- a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java @@ -133,11 +133,11 @@ public class CatalogOptions { public static final ConfigOption CACHE_MANIFEST_SIDECAR_MAX_MEMORY = key("cache.manifest-sidecar.max-memory") .memoryType() - .defaultValue(MemorySize.ofBytes(0)) + .noDefaultValue() .withDescription( "Controls the maximum memory for the separate manifest sidecar cache. " + "This budget is additional to the manifest content cache. " - + "Set to 0 to reuse the manifest content cache. It uses " + + "When unset or set to 0, reuses the manifest content cache. It uses " + "'cache.expire-after-access' and 'cache.manifest.soft-values'."); public static final ConfigOption CACHE_MANIFEST_SOFT_VALUES = diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java index 037f7ea2bfba..a4190503941c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java @@ -110,7 +110,9 @@ public CachingCatalog(Catalog wrapped, Options options) { expireAfterAccess, manifestCacheSoftValues); - MemorySize sidecarMaxMemory = options.get(CACHE_MANIFEST_SIDECAR_MAX_MEMORY); + MemorySize sidecarMaxMemory = + options.getOptional(CACHE_MANIFEST_SIDECAR_MAX_MEMORY) + .orElse(MemorySize.ofBytes(0)); this.manifestSidecarCache = sidecarMaxMemory.getBytes() == 0 ? manifestCache diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index 358ef449f8f0..2deb7512e573 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -646,8 +646,8 @@ public void testManifestCacheOptions() { public void testManifestSidecarCacheOptions() { Options options = new Options(); CachingCatalog caching = new CachingCatalog(catalog, options); - assertThat(CACHE_MANIFEST_SIDECAR_MAX_MEMORY.defaultValue()) - .isEqualTo(MemorySize.ofBytes(0)); + assertThat(CACHE_MANIFEST_SIDECAR_MAX_MEMORY.defaultValue()).isNull(); + assertThat(options.getOptional(CACHE_MANIFEST_SIDECAR_MAX_MEMORY)).isEmpty(); assertThat(caching.manifestSidecarCache).isSameAs(caching.manifestCache); options.set(CACHE_MANIFEST_SIDECAR_MAX_MEMORY, MemorySize.ofMebiBytes(8)); From 44a044cc18fb36a83f642792f932466bbe216a87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 16 Sep 2026 22:57:29 +0800 Subject: [PATCH 4/4] [core] Default manifest sidecar cache to 64 MiB --- docs/docs/concepts/spec/manifest.md | 14 +++++++------- docs/generated/catalog_configuration.html | 4 ++-- .../org/apache/paimon/options/CatalogOptions.java | 4 ++-- .../org/apache/paimon/catalog/CachingCatalog.java | 4 +--- .../apache/paimon/catalog/CachingCatalogTest.java | 10 +++++++--- 5 files changed, 19 insertions(+), 17 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index ecea86a47097..7f8bf7a8a5e7 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -80,13 +80,13 @@ Scans with partition, row-ID or bucket filters select blocks before reading mani 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 -`cache.manifest-sidecar.max-memory` (unset by default). A positive value supplies an -additional budget independent of the manifest content cache. When unset or set to 0, sidecars -reuse the manifest content cache, or remain uncached if that cache is disabled. Sidecar caching -uses the catalog's `cache.expire-after-access` and `cache.manifest.soft-values` policies. Selected block bytes -still share the manifest content cache without populating the whole-manifest entry cache -with partial results. The low-level `build` method returns sidecar bytes without writing -or publishing another file. +`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 +uses the catalog's `cache.expire-after-access` and `cache.manifest.soft-values` policies. +Selected block bytes still share the manifest content cache without populating the +whole-manifest entry cache with partial results. The low-level `build` method returns +sidecar bytes without writing or publishing another file. Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches. `build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent diff --git a/docs/generated/catalog_configuration.html b/docs/generated/catalog_configuration.html index 84c59561cdc6..bc7226d1b07d 100644 --- a/docs/generated/catalog_configuration.html +++ b/docs/generated/catalog_configuration.html @@ -52,9 +52,9 @@
cache.manifest-sidecar.max-memory
- (none) + 64 mb MemorySize - Controls the maximum memory for the separate manifest sidecar cache. This budget is additional to the manifest content cache. When unset or set to 0, reuses the manifest content cache. It uses 'cache.expire-after-access' and 'cache.manifest.soft-values'. + Controls the maximum memory for the separate manifest sidecar cache. This budget is additional to the manifest content cache. Set to 0 to reuse the manifest content cache. It uses 'cache.expire-after-access' and 'cache.manifest.soft-values'.
cache.manifest.max-memory
diff --git a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java index 09b3c934c584..bda512a944b8 100644 --- a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java @@ -133,11 +133,11 @@ public class CatalogOptions { public static final ConfigOption CACHE_MANIFEST_SIDECAR_MAX_MEMORY = key("cache.manifest-sidecar.max-memory") .memoryType() - .noDefaultValue() + .defaultValue(MemorySize.ofMebiBytes(64)) .withDescription( "Controls the maximum memory for the separate manifest sidecar cache. " + "This budget is additional to the manifest content cache. " - + "When unset or set to 0, reuses the manifest content cache. It uses " + + "Set to 0 to reuse the manifest content cache. It uses " + "'cache.expire-after-access' and 'cache.manifest.soft-values'."); public static final ConfigOption CACHE_MANIFEST_SOFT_VALUES = diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java index a4190503941c..037f7ea2bfba 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java @@ -110,9 +110,7 @@ public CachingCatalog(Catalog wrapped, Options options) { expireAfterAccess, manifestCacheSoftValues); - MemorySize sidecarMaxMemory = - options.getOptional(CACHE_MANIFEST_SIDECAR_MAX_MEMORY) - .orElse(MemorySize.ofBytes(0)); + MemorySize sidecarMaxMemory = options.get(CACHE_MANIFEST_SIDECAR_MAX_MEMORY); this.manifestSidecarCache = sidecarMaxMemory.getBytes() == 0 ? manifestCache diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index 2deb7512e573..71427641b79f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -646,9 +646,13 @@ public void testManifestCacheOptions() { public void testManifestSidecarCacheOptions() { Options options = new Options(); CachingCatalog caching = new CachingCatalog(catalog, options); - assertThat(CACHE_MANIFEST_SIDECAR_MAX_MEMORY.defaultValue()).isNull(); - assertThat(options.getOptional(CACHE_MANIFEST_SIDECAR_MAX_MEMORY)).isEmpty(); - assertThat(caching.manifestSidecarCache).isSameAs(caching.manifestCache); + assertThat(CACHE_MANIFEST_SIDECAR_MAX_MEMORY.defaultValue()) + .isEqualTo(MemorySize.ofMebiBytes(64)); + assertThat(caching.manifestSidecarCache).isNotSameAs(caching.manifestCache); + assertThat(caching.manifestSidecarCache.maxMemorySize()) + .isEqualTo(MemorySize.ofMebiBytes(64)); + assertThat(caching.manifestSidecarCache.maxElementSize()) + .isEqualTo(MemorySize.ofMebiBytes(64).getBytes()); options.set(CACHE_MANIFEST_SIDECAR_MAX_MEMORY, MemorySize.ofMebiBytes(8)); caching = new CachingCatalog(catalog, options);