This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 084cde4772de3ff587e3747f3771f75294fbf41c Author: YeJunHao <[email protected]> AuthorDate: Thu Sep 17 11:04:01 2026 +0800 [core] Prune manifest blocks with sidecars during scans (#9897) --- docs/docs/concepts/spec/manifest.md | 20 +- docs/generated/catalog_configuration.html | 6 + .../org/apache/paimon/options/CatalogOptions.java | 10 + .../benchmark/ManifestFileSorterBenchmark.java | 1 + .../java/org/apache/paimon/AbstractFileStore.java | 7 + .../src/main/java/org/apache/paimon/FileStore.java | 2 + .../org/apache/paimon/catalog/CachingCatalog.java | 21 + .../org/apache/paimon/manifest/BucketFilter.java | 16 + .../org/apache/paimon/manifest/ManifestFile.java | 85 ++- .../apache/paimon/manifest/ManifestSidecar.java | 4 +- .../paimon/operation/AbstractFileStoreScan.java | 17 +- .../paimon/table/AbstractFileStoreTable.java | 16 + .../paimon/table/DelegatedFileStoreTable.java | 11 + .../paimon/table/FallbackReadFileStoreTable.java | 6 + .../org/apache/paimon/table/FileStoreTable.java | 5 + .../java/org/apache/paimon/utils/ObjectsCache.java | 6 + .../apache/paimon/catalog/CachingCatalogTest.java | 142 +++++ .../paimon/manifest/ManifestFileMetaTest.java | 2 + .../paimon/manifest/ManifestFileMetaTestBase.java | 1 + .../apache/paimon/manifest/ManifestFileTest.java | 667 ++++++++++++++++++++- .../paimon/manifest/ManifestSidecarTest.java | 12 +- .../paimon/manifest/ManifestSidecarWriteTest.java | 1 + .../operation/ManifestRewriteCleanupTest.java | 1 + 23 files changed, 1022 insertions(+), 37 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 29466c1ad6..7f8bf7a8a5 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -72,20 +72,28 @@ 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. 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 +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 806edac3f8..bc7226d1b0 100644 --- a/docs/generated/catalog_configuration.html +++ b/docs/generated/catalog_configuration.html @@ -50,6 +50,12 @@ under the License. <td>Duration</td> <td>Cache expiration policy: marks cache entries to expire after a specified duration has passed since their last refresh.</td> </tr> + <tr> + <td><h5>cache.manifest-sidecar.max-memory</h5></td> + <td style="word-wrap: break-word;">64 mb</td> + <td>MemorySize</td> + <td>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'.</td> + </tr> <tr> <td><h5>cache.manifest.max-memory</h5></td> <td style="word-wrap: break-word;">(none)</td> 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 b1ccfe1caa..bda512a944 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<MemorySize> CACHE_MANIFEST_SIDECAR_MAX_MEMORY = + key("cache.manifest-sidecar.max-memory") + .memoryType() + .defaultValue(MemorySize.ofMebiBytes(64)) + .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<Boolean> 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 e805efb694..1f78ff8d45 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 @@ public class ManifestFileSorterBenchmark { 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 29c7ab8de0..86839355a7 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<T> implements FileStore<T> { protected final CatalogEnvironment catalogEnvironment; @Nullable private SegmentsCache<Path> readManifestCache; + @Nullable private SegmentsCache<Path> manifestSidecarCache; @Nullable private Cache<Path, Snapshot> snapshotCache; protected AbstractFileStore( @@ -212,6 +213,7 @@ abstract class AbstractFileStore<T> implements FileStore<T> { pathFactory(), options.manifestTargetSize().getBytes(), readManifestCache, + manifestSidecarCache, options); } @@ -616,6 +618,11 @@ abstract class AbstractFileStore<T> implements FileStore<T> { this.readManifestCache = manifestCache; } + @Override + public void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) { + this.manifestSidecarCache = manifestSidecarCache; + } + @Override public void setSnapshotCache(Cache<Path, Snapshot> 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 1714eec2a9..d9b143f75f 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 @@ public interface FileStore<T> { void setManifestCache(SegmentsCache<Path> manifestCache); + void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache); + void setSnapshotCache(Cache<Path, Snapshot> 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 7fce5edf0e..037f7ea2bf 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_ENABLED; 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<String, Database> databaseCache; protected Cache<Identifier, Table> tableCache; @Nullable protected final SegmentsCache<Path> manifestCache; + @Nullable protected final SegmentsCache<Path> manifestSidecarCache; // partition cache will affect data latency @Nullable protected Cache<Identifier, List<Partition>> partitionCache; @Nullable protected DVMetaCache dvMetaCache; @@ -108,6 +110,17 @@ public class CachingCatalog extends DelegateCatalog { 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 @@ public class CachingCatalog extends DelegateCatalog { if (manifestCache != null) { storeTable.setManifestCache(manifestCache); } + if (manifestSidecarCache != null) { + storeTable.setManifestSidecarCache(manifestSidecarCache); + } if (dvMetaCache != null) { storeTable.setDVMetaCache(dvMetaCache); } @@ -425,6 +441,11 @@ public class CachingCatalog extends DelegateCatalog { 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<Identifier, List<Partition>> entry : partitionCache.asMap().entrySet()) { 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 536be456ee..3e564a60f9 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 class BucketFilter { || 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 30e0adcf1b..9740961e73 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.FileUtils; 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; @@ -61,6 +62,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { private final AvroFileFormat avroFileFormat; private final long suggestedFileSize; private final CoreOptions options; + @Nullable private final SegmentsCache<Path> sidecarCache; private ManifestFile( FileIO fileIO, @@ -72,6 +74,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { PathFactory pathFactory, long suggestedFileSize, @Nullable SegmentsCache<Path> cache, + @Nullable SegmentsCache<Path> sidecarCache, CoreOptions options) { super( fileIO, @@ -89,6 +92,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { this.avroFileFormat = avroFileFormat; this.suggestedFileSize = suggestedFileSize; this.options = options; + this.sidecarCache = sidecarCache == null ? cache : sidecarCache; } @Override @@ -140,9 +144,33 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { Filter<InternalRow> readFilter, Filter<ManifestEntry> readTFilter, Function<ManifestEntry, T> convertor) { + return read( + fileName, + fileSize, + partitionFilter, + bucketFilter, + readFilter, + readTFilter, + convertor, + null); + } + + public <T> List<T> read( + String fileName, + @Nullable Long fileSize, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + Filter<InternalRow> readFilter, + Filter<ManifestEntry> readTFilter, + Function<ManifestEntry, T> 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 +183,9 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { 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 +239,23 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter) throws IOException { + return createManifestIterator( + fileIO, path, projectedType, partitionFilter, bucketFilter, null, null); + } + + private static CloseableIterator<InternalRow> createManifestIterator( + FileIO fileIO, + Path path, + RowType projectedType, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + @Nullable ManifestSidecar.Selection selected, + @Nullable SegmentsCache<Object> 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 +390,32 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { }; } + @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, + sidecarCache); + } + /** Deletes an unreferenced manifest and its explicitly referenced extra files. */ public void delete(ManifestFileMeta manifest) { delete(manifest.fileName()); @@ -365,6 +436,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { private final long suggestedFileSize; private final CoreOptions options; @Nullable private final SegmentsCache<Path> cache; + @Nullable private final SegmentsCache<Path> sidecarCache; public Factory( FileIO fileIO, @@ -375,6 +447,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { FileStorePathFactory pathFactory, long suggestedFileSize, @Nullable SegmentsCache<Path> cache, + @Nullable SegmentsCache<Path> sidecarCache, CoreOptions options) { this.fileIO = fileIO; this.schemaManager = schemaManager; @@ -384,13 +457,10 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { 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, @@ -402,6 +472,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { 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 c6ee9e9f56..6a8ce2b143 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 final class ManifestSidecar { @Nullable PartitionPredicate partitionFilter, @Nullable RowType partitionType, @Nullable BiPredicate<Integer, Integer> bucketFilter, - @Nullable SegmentsCache<Object> cache) { + @Nullable SegmentsCache<Path> cache) { String sidecarFileName = fileName(manifest); if (sidecarFileName == null) { return null; @@ -660,7 +660,7 @@ public final class ManifestSidecar { } } - /** 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/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java index 82923ef9d8..e40d27e789 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.ManifestEntry; 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 @@ public abstract class AbstractFileStoreScan implements FileStoreScan { @Nullable Filter<InternalRow> additionalFilter, @Nullable Filter<ManifestEntry> 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<InternalRow> entryRowFilter = createEntryRowFilter(); Function<ManifestEntry, T> finalConverter = dropStats ? e -> converter.apply(dropStats(e)) : converter; List<T> 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/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index e9d9603f5a..93ecc173d5 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<String> appliedDynamicOptionKeys; @Nullable protected transient SegmentsCache<Path> manifestCache; + @Nullable protected transient SegmentsCache<Path> manifestSidecarCache; @Nullable protected transient Cache<Path, Snapshot> snapshotCache; @Nullable protected transient Cache<String, Statistics> statsCache; @Nullable protected transient DVMetaCache dvmetaCache; @@ -144,6 +145,18 @@ abstract class AbstractFileStoreTable implements FileStoreTable { return manifestCache; } + @Override + public void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) { + this.manifestSidecarCache = manifestSidecarCache; + store().setManifestSidecarCache(manifestSidecarCache); + } + + @Nullable + @Override + public SegmentsCache<Path> getManifestSidecarCache() { + return manifestSidecarCache; + } + @Override public void setSnapshotCache(Cache<Path, Snapshot> cache) { this.snapshotCache = cache; @@ -432,6 +445,9 @@ abstract class AbstractFileStoreTable implements FileStoreTable { 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 9a5e81bcb6..ed541721fc 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 abstract class DelegatedFileStoreTable implements FileStoreTable { return wrapped.getManifestCache(); } + @Override + public void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) { + wrapped.setManifestSidecarCache(manifestSidecarCache); + } + + @Nullable + @Override + public SegmentsCache<Path> getManifestSidecarCache() { + return wrapped.getManifestSidecarCache(); + } + @Override public void setSnapshotCache(Cache<Path, Snapshot> 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 da37e07808..940dd38b2b 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 class FallbackReadFileStoreTable extends DelegatedFileStoreTable { other.setManifestCache(manifestCache); } + @Override + public void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) { + super.setManifestSidecarCache(manifestSidecarCache); + other.setManifestSidecarCache(manifestSidecarCache); + } + protected FileStoreTable switchWrappedToBranch(String branchName) { Optional<TableSchema> 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 9704934305..ddaa174076 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 @@ public interface FileStoreTable extends DataTable { @Nullable SegmentsCache<Path> getManifestCache(); + void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache); + + @Nullable + SegmentsCache<Path> getManifestSidecarCache(); + void setSnapshotCache(Cache<Path, Snapshot> cache); void setStatsCache(Cache<String, Statistics> cache); 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 4530f698e2..59106c5a2a 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 abstract class ObjectsCache<K, V, S extends Segments> { this.cacheMetrics = cacheMetrics; } + /** Shares the byte cache with consumers using distinct whole-file and block keys. */ + @SuppressWarnings("unchecked") + public SegmentsCache<Object> segmentsCache() { + return (SegmentsCache<Object>) (SegmentsCache<?>) cache; + } + public List<V> read(K key, @Nullable Long fileSize, Filters<V> filters) throws IOException { return read(key, fileSize, filters, Function.identity()); } 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 3de7bbf73f..71427641b7 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.CoreOptions; 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.apache.paimon.shade.guava30.com.google.common.collect.Lists; 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 java.util.Collections.singletonMap; 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,138 @@ class CachingCatalogTest extends CatalogTestBase { 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.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); + 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 16f33f4cee..161591a48e 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 class ManifestFileMetaTest extends ManifestFileMetaTestBase { null), Long.MAX_VALUE, null, + null, new CoreOptions(new Options())) .create(); @@ -3206,6 +3207,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { 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 a4516015c2..4e72f58e73 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 @@ public abstract class ManifestFileMetaTestBase { 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 0362871ef8..db914239ee 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.FileIO; 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.CloseableIterator; 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.List; 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,16 +1418,646 @@ public class ManifestFileTest { return -1; } + @Test + 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<Path> cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); + SegmentsCache<Path> sidecarCache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); + ManifestFile.Factory factory = + createManifestFileFactory( + tempDir.toString(), Long.MAX_VALUE, options, io, cache, sidecarCache); + List<ManifestEntry> 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(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))); + assertThat(factory.create().selectBlocks(meta, hit).blocks()).isNotEmpty(); + assertThat(io.opened).isEmpty(); + assertThat(factory.create().read(meta.fileName())) + .containsExactlyInAnyOrderElementsOf(entries); + 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(); + assertThat(factory.create().read(meta.fileName())) + .containsExactlyInAnyOrderElementsOf(entries); + 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<Path> manifestCache = + cacheManifest + ? new SegmentsCache<>( + 1024, + MemorySize.ofMebiBytes(1), + Long.MAX_VALUE, + null, + false) + : null; + SegmentsCache<Path> 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(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + RecordingFileIO fileIO = new RecordingFileIO(); + SegmentsCache<Path> 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<ManifestEntry> 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<ManifestEntry> actual = readSelectedEntries(factory.create(), meta, selected); + List<ManifestEntry> 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<Path> 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<ManifestFileMeta> 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<ManifestEntry> 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<ManifestEntry> 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<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 = 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<ManifestFileMeta> 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<ManifestEntry> 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<ManifestEntry> 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<String> extraFiles : + Arrays.asList( + null, + Collections.<String>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<ManifestEntry> 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<Path> opened = Collections.synchronizedList(new ArrayList<>()); + private final List<Long> 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<Path> 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<Path> cache) { + return createManifestFileFactory(pathStr, suggestedFileSize, options, fileIO, cache, null); + } + + private ManifestFile.Factory createManifestFileFactory( + String pathStr, + long suggestedFileSize, + Options options, + FileIO fileIO, + @Nullable SegmentsCache<Path> cache, + @Nullable SegmentsCache<Path> sidecarCache) { Path path = new Path(pathStr); FileStorePathFactory pathFactory = new FileStorePathFactory( @@ -1433,18 +2076,18 @@ public class ManifestFileTest { 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), + DEFAULT_PART_TYPE, + avro, + "zstd", + pathFactory, + suggestedFileSize, + cache, + sidecarCache, + coreOptions); } @Test 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 1110c70158..1b3f9d258e 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 @@ class ManifestSidecarTest { Files.write(temp.resolve(sidecar.getName()), data); ManifestFileMeta meta = testMeta(); FileIO io = spy(LocalFileIO.create()); - SegmentsCache<Object> tooSmall = + SegmentsCache<Path> 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<Object> cache = + SegmentsCache<Path> 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 @@ class ManifestSidecarTest { Path path = new Path(temp.toString(), "manifest-golden"); Path sidecar = ManifestSidecar.path(path); Files.write(temp.resolve(sidecar.getName()), data); - SegmentsCache<Object> cache = + SegmentsCache<Path> 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 @@ class ManifestSidecarTest { Path sidecar = ManifestSidecar.path(path); ManifestFileMeta meta = testMeta(); FileIO io = spy(LocalFileIO.create()); - SegmentsCache<Object> cache = + SegmentsCache<Path> 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 @@ class ManifestSidecarTest { Files.write(temp.resolve(sidecar.getName()), testSidecar()); ManifestFileMeta meta = testMeta(); FileIO io = spy(LocalFileIO.create()); - SegmentsCache<Object> cache = + SegmentsCache<Path> 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 @@ class ManifestSidecarTest { } private ManifestSidecar.Selection readCached( - FileIO io, Path path, ManifestFileMeta meta, SegmentsCache<Object> cache) { + FileIO io, Path path, ManifestFileMeta meta, SegmentsCache<Path> 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 cea952943f..3549bf6dcd 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 @@ class ManifestSidecarWriteTest { 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 20a1b26bee..dd1a1b176b 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 @@ class ManifestRewriteCleanupTest extends ManifestFileMetaTestBase { null), suggestedFileSize, null, + null, new CoreOptions(new Options())) .create(); }
