diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 5b4271cc215e..24733b51959d 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -84,9 +84,16 @@ perform sidecar I/O. Sidecar caching is controlled by the catalog option 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. +Manifest contents use one decoded-entry cache keyed by physical block path, offset and length. +Full reads and sidecar-selected reads reuse the same complete, unfiltered blocks; compressed +block bytes and whole-file entry copies are not retained separately. Query filters and +converters run after loading the cached entries. Decoded buffers and their lookup metadata +are accounted against the manifest content cache budget, and oversized blocks are read +without retaining a partial cache entry. A complete block directory may be cached separately +from the entries after reaching EOF, or after selecting every physical block. Without that +directory, the first full read still scans the manifest to discover block boundaries, while +reusing any already decoded blocks. The low-level `build` method returns sidecar bytes without +writing or publishing another file. PyPaimon can read these sidecars and prune manifest blocks using partition, row-ID and bucket filters. Its `manifest.sidecar.enabled` option inherits `manifest-sort.enabled` when unset. @@ -305,17 +312,21 @@ not sidecar storage I/O. Selected compressed Avro blocks are read by byte range spans coalesced and individual read requests bounded to 4 MiB. Building a sidecar does not modify the original manifest. -`read` and `openManifest` accept an optional caller-supplied `SegmentsCache`. Complete -sidecar bytes are keyed by their explicit `Path`. Only successful reads and selections +`read` accepts an optional caller-supplied `SegmentsCache`. Complete sidecar bytes +are keyed by their explicit `Path`. Only successful reads and selections populate the cache; query-specific selections are not cached. Cache entry-size limits affect admission only: larger sidecars are still fully read, validated and used. -Selected Avro blocks also share this cache. Each entry contains one complete compressed block, -keyed by the manifest's full path, original offset and encoded length, separately from whole-file -keys. Different selections reuse the same blocks. Only complete reads populate the cache; -oversized blocks stream through the read buffer. Adjacent uncached blocks fitting the buffer -are read together and cached individually. Fully cached selections do not open the manifest. -The cache retains its configured memory budget, entry-size limit, expiration and eviction policy. +Manifest reads cache the decoded entries of each complete Avro block, keyed by +`(path, offset, length)` in a namespace distinct from metadata path keys. These entries have +the complete manifest schema and are independent of query filters and converters. Full and +selected reads reuse the same entries; there is no separate compressed-block cache. Only +successful complete decoding populates a block entry. If decoded buffers exceed the entry +or memory limit, the reader replays its current raw block without another file read and +without retaining a filtered prefix. Adjacent missing blocks are fetched together in bounded +reads. Fully cached selections do not open the manifest. A path-keyed complete block directory +contains only the header and physical descriptors, not a second copy of manifest entries. +The caches retain their configured memory budgets, expiration and eviction policies. ## Manifest diff --git a/docs/generated/catalog_configuration.html b/docs/generated/catalog_configuration.html index bc7226d1b07d..a9a3df0456c5 100644 --- a/docs/generated/catalog_configuration.html +++ b/docs/generated/catalog_configuration.html @@ -72,7 +72,7 @@
cache.manifest.small-file-threshold
1 mb MemorySize - Controls the threshold of small manifest file. + Controls the per-element cache threshold for metadata files and decoded manifest blocks.
cache.manifest.soft-values
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 bda512a944b8..5334cfcfdf3c 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 @@ -122,7 +122,8 @@ public class CatalogOptions { key("cache.manifest.small-file-threshold") .memoryType() .defaultValue(MemorySize.ofMebiBytes(1)) - .withDescription("Controls the threshold of small manifest file."); + .withDescription( + "Controls the per-element cache threshold for metadata files and decoded manifest blocks."); public static final ConfigOption CACHE_MANIFEST_MAX_MEMORY = key("cache.manifest.max-memory") diff --git a/paimon-common/src/main/java/org/apache/paimon/io/DataPagedOutputSerializer.java b/paimon-common/src/main/java/org/apache/paimon/io/DataPagedOutputSerializer.java index c7ce3c94ec8f..0b591d5dc075 100644 --- a/paimon-common/src/main/java/org/apache/paimon/io/DataPagedOutputSerializer.java +++ b/paimon-common/src/main/java/org/apache/paimon/io/DataPagedOutputSerializer.java @@ -76,6 +76,13 @@ SimpleCollectingOutputView pagedOut() { return pagedOut; } + /** Returns the bytes currently allocated for serialized data. */ + public long memorySize() { + return pagedOut == null + ? initialOut.getSharedBuffer().length + : (long) pagedOut.fullSegments().size() * pageSize; + } + /** * Serializes a binary row to the output. * diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/BlockKey.java b/paimon-core/src/main/java/org/apache/paimon/manifest/BlockKey.java new file mode 100644 index 000000000000..1549aeed6ceb --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/BlockKey.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.manifest; + +import org.apache.paimon.fs.Path; + +import java.util.Objects; + +/** Identifies one complete physical block of an immutable manifest file. */ +public final class BlockKey { + + private final Path path; + private final long offset; + private final long length; + + public BlockKey(Path path, long offset, long length) { + this.path = Objects.requireNonNull(path); + this.offset = offset; + this.length = length; + } + + public Path path() { + return path; + } + + public long offset() { + return offset; + } + + public long length() { + return length; + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof BlockKey)) { + return false; + } + BlockKey that = (BlockKey) other; + return path.equals(that.path) && offset == that.offset && length == that.length; + } + + @Override + public int hashCode() { + return Objects.hash(path, offset, length); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java index 95134f36a002..742ccbfaf29c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java @@ -439,7 +439,7 @@ public RowIterator toRows(RowType projectedType) throws IOException { return toRows(projectedType, null, null, true); } - private RowIterator toRows( + RowIterator toRows( RowType projectedType, @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter, diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestBlockReader.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestBlockReader.java new file mode 100644 index 000000000000..12a73a4938fc --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestBlockReader.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.manifest; + +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.FileUtils; + +import javax.annotation.Nullable; + +import java.io.Closeable; +import java.io.EOFException; +import java.io.IOException; + +/** Query-local reader of full manifests or a sequence of cache-missing physical blocks. */ +final class ManifestBlockReader implements Closeable { + + private final FileIO fileIO; + private final Path path; + @Nullable private final ManifestSidecar.Selection selected; + @Nullable private ManifestAvroReader reader; + @Nullable private ManifestAvroReader.RawBlock current; + @Nullable private BlockKey currentKey; + private int position; + private long firstRecord; + + ManifestBlockReader(FileIO fileIO, Path path, @Nullable ManifestSidecar.Selection selected) { + this.fileIO = fileIO; + this.path = path; + this.selected = selected; + } + + private ManifestAvroReader reader() throws IOException { + if (reader == null) { + try { + reader = + new ManifestAvroReader( + ManifestSidecar.openManifest(fileIO, path, selected)); + } catch (IOException e) { + FileUtils.checkExists(fileIO, path); + throw e; + } + } + return reader; + } + + byte[] header() throws IOException { + return reader().headerBytes(); + } + + boolean hasNext() throws IOException { + return reader().hasNext(); + } + + ManifestSidecar.Block next() throws IOException { + ManifestAvroReader input = reader(); + current = input.next(); + ManifestSidecar.Block block; + if (selected == null) { + block = + new ManifestSidecar.Block( + input.blockOffset(), + input.blockLength(), + firstRecord, + current.recordCount()); + } else { + if (position >= selected.blocks().size()) { + throw new IOException("Unexpected manifest block"); + } + block = selected.blocks().get(position++); + if (current.recordCount() != block.recordCount) { + throw new IOException("Manifest block record count does not match its directory"); + } + } + firstRecord = Math.addExact(firstRecord, current.recordCount()); + currentKey = new BlockKey(path, block.offset, block.length); + return block; + } + + CloseableIterator rows( + BlockKey key, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter) + throws IOException { + while (currentKey == null || currentKey.offset() < key.offset()) { + if (!hasNext()) { + throw new EOFException("Missing manifest block at " + key.offset()); + } + next(); + } + if (!key.equals(currentKey)) { + throw new IOException("Unexpected manifest block position"); + } + ManifestAvroReader.RowIterator rows = + current.toRows( + ManifestEntry.MANIFEST_ROW_TYPE, partitionFilter, bucketFilter, true); + return new CloseableIterator() { + + @Override + public boolean hasNext() { + return rows.hasNext(); + } + + @Override + public InternalRow next() { + return rows.next(); + } + + @Override + public void close() { + // The enclosing query owns the file reader, not an individual block iterator. + } + }; + } + + @Override + public void close() throws IOException { + if (reader != null) { + reader.close(); + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java index ab5b87551e40..1e281f332c3c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java @@ -23,7 +23,6 @@ import org.apache.paimon.data.Segments; import org.apache.paimon.data.SimpleCollectingOutputView; import org.apache.paimon.data.serializer.InternalRowSerializer; -import org.apache.paimon.fs.Path; import org.apache.paimon.io.DataPagedOutputSerializer; import org.apache.paimon.manifest.ManifestEntrySegments.RichSegments; import org.apache.paimon.partition.PartitionPredicate; @@ -44,14 +43,11 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.OptionalInt; import java.util.function.Function; -import java.util.function.Supplier; -import java.util.stream.Collectors; import static org.apache.paimon.manifest.ManifestEntrySerializer.bucketGetter; import static org.apache.paimon.manifest.ManifestEntrySerializer.partitionGetter; @@ -63,25 +59,26 @@ * ManifestEntrySegments}. */ @ThreadSafe -public class ManifestEntryCache extends ObjectsCache { +public class ManifestEntryCache + extends ObjectsCache { @Nullable private final FilteredReader filteredReader; public ManifestEntryCache( - SegmentsCache cache, + SegmentsCache cache, ObjectSerializer projectedSerializer, RowType formatSchema, - FunctionWithIOException fileSizeFunction, - BiFunctionWithIOE> reader) { + FunctionWithIOException fileSizeFunction, + BiFunctionWithIOE> reader) { this(cache, projectedSerializer, formatSchema, fileSizeFunction, reader, null); } public ManifestEntryCache( - SegmentsCache cache, + SegmentsCache cache, ObjectSerializer projectedSerializer, RowType formatSchema, - FunctionWithIOException fileSizeFunction, - BiFunctionWithIOE> reader, + FunctionWithIOException fileSizeFunction, + BiFunctionWithIOE> reader, @Nullable FilteredReader filteredReader) { super(cache, projectedSerializer, formatSchema, fileSizeFunction, reader); this.filteredReader = filteredReader; @@ -90,20 +87,21 @@ public ManifestEntryCache( /** Uncached reads skip non-matching partitions and buckets before decoding their stats. */ @Override protected CloseableIterator createFilteredIterator( - Path path, @Nullable Long fileSize, Filters filters) throws IOException { + BlockKey key, @Nullable Long fileSize, Filters filters) + throws IOException { if (filteredReader != null && filters instanceof ManifestEntryFilters) { ManifestEntryFilters manifestFilters = (ManifestEntryFilters) filters; return filteredReader.read( - path, fileSize, manifestFilters.partitionFilter, manifestFilters.bucketFilter); + key, fileSize, manifestFilters.partitionFilter, manifestFilters.bucketFilter); } - return super.createFilteredIterator(path, fileSize, filters); + return super.createFilteredIterator(key, fileSize, filters); } /** Reader of manifest rows which can skip entries by partition and bucket while decoding. */ @FunctionalInterface public interface FilteredReader { CloseableIterator read( - Path path, + BlockKey key, @Nullable Long fileSize, @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter) @@ -111,42 +109,103 @@ CloseableIterator read( } @Override - protected ManifestEntrySegments createSegments(Path path, @Nullable Long fileSize) { - Map, DataPagedOutputSerializer> segments = - new HashMap<>(); + protected ManifestEntrySegments createSegments(BlockKey key, @Nullable Long fileSize) { + List segments = new ArrayList<>(); Function partitionGetter = partitionGetter(); Function bucketGetter = bucketGetter(); Function totalBucketGetter = totalBucketGetter(); int pageSize = cache.pageSize(); InternalRowSerializer formatSerializer = this.formatSerializer.get(); - Supplier outputSupplier = - () -> new DataPagedOutputSerializer(formatSerializer, 2048, pageSize); - try (CloseableIterator iterator = reader.apply(path, fileSize)) { + Triple group = null; + DataPagedOutputSerializer output = null; + long completedBytes = 0; + long limit = Math.min(cache.maxElementSize(), cache.maxMemorySize().getBytes()); + try (CloseableIterator iterator = reader.apply(key, fileSize)) { while (iterator.hasNext()) { InternalRow row = iterator.next(); BinaryRow partition = partitionGetter.apply(row); int bucket = bucketGetter.apply(row); int totalBucket = totalBucketGetter.apply(row); - Triple key = Triple.of(partition, bucket, totalBucket); - DataPagedOutputSerializer output = - segments.computeIfAbsent(key, k -> outputSupplier.get()); + // Keep consecutive runs rather than regrouping the block: physical entry order + // must survive cache hits, including ADD/DELETE entries and overlapping row IDs. + if (group == null + || !group.f0.equals(partition) + || group.f1 != bucket + || group.f2 != totalBucket) { + if (output != null) { + RichSegments completed = finish(group, output); + segments.add(completed); + completedBytes += completed.totalMemorySize(); + } + group = Triple.of(partition.copy(), bucket, totalBucket); + output = new DataPagedOutputSerializer(formatSerializer, 2048, pageSize); + } output.write(row); + if (completedBytes + output.memorySize() + RichSegments.metadataMemorySize(group.f0) + > limit) { + throw new CacheLimitExceeded(); + } } - List result = new ArrayList<>(); - for (Map.Entry, DataPagedOutputSerializer> entry : - segments.entrySet()) { - Triple key = entry.getKey(); - SimpleCollectingOutputView view = entry.getValue().close(); - Segments seg = - Segments.create(view.fullSegments(), view.getCurrentPositionInSegment()); - result.add(new RichSegments(key.f0, key.f1, key.f2, seg)); + if (output != null) { + segments.add(finish(group, output)); } - return new ManifestEntrySegments(result); + return new ManifestEntrySegments(segments); + } catch (CacheLimitExceeded e) { + throw e; } catch (Exception e) { throw new RuntimeException(e); } } + private static RichSegments finish( + Triple group, DataPagedOutputSerializer output) + throws IOException { + SimpleCollectingOutputView view = output.close(); + Segments data = Segments.create(view.fullSegments(), view.getCurrentPositionInSegment()); + return new RichSegments(group.f0, group.f1, group.f2, data); + } + + List readCached( + ManifestEntrySegments entries, + Filters filters, + Function convertor) + throws IOException { + if (cacheMetrics != null) { + cacheMetrics.increaseHitObject(); + } + return readFromSegments(entries, filters, convertor); + } + + @Override + public List read( + BlockKey key, + @Nullable Long fileSize, + Filters filters, + Function convertor) + throws IOException { + try { + return super.read(key, fileSize, filters, convertor); + } catch (CacheLimitExceeded ignored) { + // The query-local reader can replay its current raw block without another file read. + // Do not cache a filtered prefix of a block that exceeded the decoded-memory limit. + return org.apache.paimon.utils.ObjectsFile.readFromIterator( + createFilteredIterator(key, fileSize, filters), + projectedSerializer, + filters.readFilter(), + filters.readVFilter(), + convertor); + } + } + + private static final class CacheLimitExceeded extends RuntimeException { + + private static final long serialVersionUID = 1L; + + private CacheLimitExceeded() { + super(null, null, false, false); + } + } + @Override protected List readFromSegments( ManifestEntrySegments manifestSegments, @@ -176,9 +235,6 @@ protected List readFromSegments( if (segments == null) { return Collections.emptyList(); } - } else { - segments = - segMap.values().stream().flatMap(List::stream).collect(Collectors.toList()); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySegments.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySegments.java index 212d9ab8d2b3..dd9bcbed7638 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySegments.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySegments.java @@ -22,6 +22,7 @@ import org.apache.paimon.data.Segments; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -38,12 +39,8 @@ public class ManifestEntrySegments implements Segments { private final Map>> indexedSegments; public ManifestEntrySegments(List segments) { - this.segments = segments; - this.totalMemorySize = - segments.stream() - .map(RichSegments::segments) - .mapToLong(Segments::totalMemorySize) - .sum(); + this.segments = Collections.unmodifiableList(new ArrayList<>(segments)); + this.totalMemorySize = segments.stream().mapToLong(RichSegments::totalMemorySize).sum(); this.indexedSegments = new HashMap<>(); for (RichSegments seg : segments) { indexedSegments @@ -96,5 +93,14 @@ public int totalBucket() { public Segments segments() { return segments; } + + long totalMemorySize() { + // Include the partition copy and an estimate for the run and lookup-index objects. + return segments.totalMemorySize() + metadataMemorySize(partition); + } + + static long metadataMemorySize(BinaryRow partition) { + return partition.getSizeInBytes() + 192L; + } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java index 9740961e73d2..b5b86261c532 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 @@ -21,6 +21,7 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.Segments; import org.apache.paimon.format.FileFormat; import org.apache.paimon.format.avro.AvroFileFormat; import org.apache.paimon.fs.FileIO; @@ -46,6 +47,7 @@ import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.function.Function; @@ -62,7 +64,9 @@ public class ManifestFile extends ObjectsFile { private final AvroFileFormat avroFileFormat; private final long suggestedFileSize; private final CoreOptions options; + @Nullable private final SegmentsCache manifestCache; @Nullable private final SegmentsCache sidecarCache; + @Nullable private CacheMetrics cacheMetrics; private ManifestFile( FileIO fileIO, @@ -86,37 +90,32 @@ private ManifestFile( avroFileFormat.createWriterFactory(ManifestEntry.MANIFEST_ROW_TYPE), compression, pathFactory, - cache); + null); this.schemaManager = schemaManager; this.partitionType = partitionType; this.avroFileFormat = avroFileFormat; this.suggestedFileSize = suggestedFileSize; this.options = options; + this.manifestCache = cache; this.sidecarCache = sidecarCache == null ? cache : sidecarCache; } @Override - protected ManifestEntryCache createCache( - @Nullable SegmentsCache cache, RowType formatType) { - return new ManifestEntryCache( - cache, - serializer, - formatType, - super::fileSize, - this::createIterator, - (path, fileSize, partitionFilter, bucketFilter) -> - createManifestIterator( - fileIO, - path, - ManifestEntry.MANIFEST_ROW_TYPE, - partitionFilter, - bucketFilter)); + public ManifestFile withCacheMetrics(@Nullable CacheMetrics cacheMetrics) { + this.cacheMetrics = cacheMetrics; + return this; } @Override - public ManifestFile withCacheMetrics(@Nullable CacheMetrics cacheMetrics) { - super.withCacheMetrics(cacheMetrics); - return this; + protected List readWithIOException( + String fileName, + @Nullable Long fileSize, + Filter readFilter, + Filter readTFilter, + Function convertor) + throws IOException { + return readEntries( + fileName, fileSize, null, null, readFilter, readTFilter, convertor, null); } public List read( @@ -164,31 +163,195 @@ public List read( 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); - // 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); - return cache.read(path, fileSize, filters, convertor); - } + return readEntries( + fileName, + fileSize, + partitionFilter, + bucketFilter, + readFilter, + readTFilter, + convertor, + selected); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } - CloseableIterator iterator = + private List readEntries( + String fileName, + @Nullable Long fileSize, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + Filter readFilter, + Filter readTFilter, + Function convertor, + @Nullable ManifestSidecar.Selection selected) + throws IOException { + if (selected != null && selected.blocks().isEmpty()) { + return Collections.emptyList(); + } + Path path = pathFactory.toPath(fileName); + if (manifestCache == null) { + return readFromIterator( createManifestIterator( fileIO, path, ManifestEntry.MANIFEST_ROW_TYPE, partitionFilter, bucketFilter, - selected, - cache == null ? null : cache.segmentsCache()); - return readFromIterator(iterator, serializer, readFilter, readTFilter, convertor); - } catch (IOException e) { - throw new UncheckedIOException(e); + selected), + serializer, + readFilter, + readTFilter, + convertor); + } + ManifestEntryFilters filters = + new ManifestEntryFilters(partitionFilter, bucketFilter, readFilter, readTFilter); + if (selected == null) { + Segments cached = manifestCache.getIfPresents(path); + if (cached instanceof BlockDirectory) { + BlockDirectory directory = (BlockDirectory) cached; + if (fileSize == null || fileSize == directory.fileSize) { + selected = directory.selection; + } + } + } + return selected == null + ? readAllBlocks(path, fileSize, filters, convertor) + : readSelectedBlocks(path, fileSize, selected, filters, convertor); + } + + @SuppressWarnings("unchecked") + private SegmentsCache blockCache() { + // The catalog budget is shared by path-keyed metadata and physical block keys. + return (SegmentsCache) (SegmentsCache) manifestCache; + } + + private ManifestEntryCache entryCache(ManifestBlockReader input) { + ManifestEntryCache entries = + new ManifestEntryCache( + blockCache(), + serializer, + ManifestEntry.MANIFEST_ROW_TYPE, + BlockKey::length, + (key, size) -> input.rows(key, null, null), + (key, size, partition, bucket) -> input.rows(key, partition, bucket)); + entries.withCacheMetrics(cacheMetrics); + return entries; + } + + private List readAllBlocks( + Path path, + @Nullable Long fileSize, + ManifestEntryFilters filters, + Function convertor) + throws IOException { + List result = new ArrayList<>(); + List blocks = new ArrayList<>(); + byte[] header; + long size; + try (ManifestBlockReader input = new ManifestBlockReader(fileIO, path, null)) { + header = input.header(); + size = header.length; + ManifestEntryCache entries = entryCache(input); + while (input.hasNext()) { + ManifestSidecar.Block block = input.next(); + blocks.add(block); + size = Math.addExact(block.offset, block.length); + result.addAll( + entries.read( + new BlockKey(path, block.offset, block.length), + block.length, + filters, + convertor)); + } + if (fileSize != null && fileSize != size) { + throw new IOException("Manifest size does not match its block directory"); + } + } + // Only EOF proves that this directory describes the entire file. + cacheDirectory(path, size, new ManifestSidecar.Selection(header, blocks)); + return result; + } + + private List readSelectedBlocks( + Path path, + @Nullable Long fileSize, + ManifestSidecar.Selection selected, + ManifestEntryFilters filters, + Function convertor) + throws IOException { + List hits = new ArrayList<>(); + List misses = new ArrayList<>(); + SegmentsCache cache = blockCache(); + for (ManifestSidecar.Block block : selected.blocks()) { + Segments value = cache.getIfPresents(new BlockKey(path, block.offset, block.length)); + ManifestEntrySegments entries = + value instanceof ManifestEntrySegments ? (ManifestEntrySegments) value : null; + // Pin hits for this read, so concurrent eviction cannot invalidate the miss sequence. + hits.add(entries); + if (entries == null) { + misses.add(block); + } + } + List result = new ArrayList<>(); + try (ManifestBlockReader input = + new ManifestBlockReader( + fileIO, path, new ManifestSidecar.Selection(selected.header(), misses))) { + ManifestEntryCache entries = entryCache(input); + for (int i = 0; i < selected.blocks().size(); i++) { + ManifestSidecar.Block block = selected.blocks().get(i); + ManifestEntrySegments hit = hits.get(i); + result.addAll( + hit == null + ? entries.read( + new BlockKey(path, block.offset, block.length), + block.length, + filters, + convertor) + : entries.readCached(hit, filters, convertor)); + } + } + if (fileSize != null && isComplete(selected, fileSize)) { + cacheDirectory(path, fileSize, selected); + } + return result; + } + + private static boolean isComplete(ManifestSidecar.Selection selected, long fileSize) { + long offset = selected.header().length; + for (ManifestSidecar.Block block : selected.blocks()) { + if (block.offset != offset || block.length > fileSize - offset) { + return false; + } + offset += block.length; + } + return offset == fileSize; + } + + private void cacheDirectory(Path path, long fileSize, ManifestSidecar.Selection selected) { + BlockDirectory directory = new BlockDirectory(fileSize, selected); + if (directory.totalMemorySize() <= manifestCache.maxElementSize() + && directory.totalMemorySize() <= manifestCache.maxMemorySize().getBytes()) { + manifestCache.put(path, directory); + } + } + + /** Complete physical directory only; no duplicate copy of the entries is retained here. */ + private static final class BlockDirectory implements Segments { + + private final long fileSize; + private final ManifestSidecar.Selection selection; + + private BlockDirectory(long fileSize, ManifestSidecar.Selection selection) { + this.fileSize = fileSize; + this.selection = selection; + } + + @Override + public long totalMemorySize() { + return selection.header().length + 64L * selection.blocks().size() + 64; } } @@ -240,7 +403,7 @@ private static CloseableIterator createManifestIterator( @Nullable BucketFilter bucketFilter) throws IOException { return createManifestIterator( - fileIO, path, projectedType, partitionFilter, bucketFilter, null, null); + fileIO, path, projectedType, partitionFilter, bucketFilter, null); } private static CloseableIterator createManifestIterator( @@ -249,13 +412,11 @@ private static CloseableIterator createManifestIterator( RowType projectedType, @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter, - @Nullable ManifestSidecar.Selection selected, - @Nullable SegmentsCache cache) + @Nullable ManifestSidecar.Selection selected) throws IOException { try { ManifestAvroReader reader = - new ManifestAvroReader( - ManifestSidecar.openManifest(fileIO, path, selected, cache)); + new ManifestAvroReader(ManifestSidecar.openManifest(fileIO, path, selected)); return reader.read(projectedType, partitionFilter, bucketFilter); } catch (IOException e) { FileUtils.checkExists(fileIO, path); diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java index 6a8ce2b143bb..cb3867eb716e 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 @@ -46,7 +46,6 @@ import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; @@ -112,12 +111,17 @@ public Block(long offset, long length, long firstRecord, long recordCount) { /** Selected blocks in original file order. Empty means the manifest can be excluded. */ public static final class Selection { + private final byte[] header; private final List blocks; - private Selection(byte[] header, List blocks) { + Selection(byte[] header, List blocks) { this.header = header; - this.blocks = Collections.unmodifiableList(blocks); + this.blocks = Collections.unmodifiableList(new ArrayList<>(blocks)); + } + + byte[] header() { + return header; } public List blocks() { @@ -692,67 +696,17 @@ private static byte[] readBytes(FileIO io, Path path) throws IOException { static InputStream openManifest(FileIO io, Path path, @Nullable Selection selected) throws IOException { - return openManifest(io, path, selected, null); - } - - static InputStream openManifest( - FileIO io, - Path path, - @Nullable Selection selected, - @Nullable SegmentsCache cache) - throws IOException { return selected == null ? io.newInputStream(path) - : new SelectedBlockInput(io, path, selected, cache); - } - - /** Separates physical byte ranges from whole-file cache keys. */ - static final class BlockCacheKey { - private final Path path; - private final long offset; - private final long length; - - BlockCacheKey(Path path, long offset, long length) { - this.path = Objects.requireNonNull(path); - this.offset = offset; - this.length = length; - } - - @Override - public boolean equals(Object other) { - if (!(other instanceof BlockCacheKey)) { - return false; - } - BlockCacheKey that = (BlockCacheKey) other; - return path.equals(that.path) && offset == that.offset && length == that.length; - } - - @Override - public int hashCode() { - return Objects.hash(path, offset, length); - } - } - - /** Complete encoded Avro blocks, distinct from cached manifest entries and sidecar bytes. */ - private static final class ManifestBlockSegment implements Segments { - private final byte[] bytes; - - private ManifestBlockSegment(byte[] bytes) { - this.bytes = bytes; - } - - @Override - public long totalMemorySize() { - return bytes.length; - } + : new SelectedBlockInput(io, path, selected); } /** An OCF stream comprising the original header and selected complete compressed blocks. */ private static final class SelectedBlockInput extends InputStream { + private final FileIO io; private final Path path; private final Selection selected; - @Nullable private final SegmentsCache cache; @Nullable private SeekableInputStream input; private boolean closed; private int headerPosition; @@ -762,12 +716,10 @@ private static final class SelectedBlockInput extends InputStream { private int bufferPosition; private int bufferLimit; - private SelectedBlockInput( - FileIO io, Path path, Selection selected, @Nullable SegmentsCache cache) { + private SelectedBlockInput(FileIO io, Path path, Selection selected) { this.io = io; this.path = path; this.selected = selected; - this.cache = cache; } @Override @@ -808,26 +760,17 @@ private boolean fillBuffer() throws IOException { if (blockPosition == selected.blocks.size()) { return false; } - Block next = selected.blocks.get(blockPosition); - if (cache != null && next.length <= cache.maxElementSize()) { - readCachedBlocks(next); - return true; - } Block block = selected.blocks.get(blockPosition++); long end = block.offset + block.length; while (blockPosition < selected.blocks.size() - && selected.blocks.get(blockPosition).offset == end - && (cache == null - || selected.blocks.get(blockPosition).length - > cache.maxElementSize())) { + && selected.blocks.get(blockPosition).offset == end) { end += selected.blocks.get(blockPosition++).length; } seekInput(block.offset); remaining = end - block.offset; } int requested = (int) Math.min(BLOCK_READ_BUFFER_BYTES, remaining); - // A previous buffer may be shared with other readers through the block cache. - if (cache != null || buffer == null || buffer.length < requested) { + if (buffer == null || buffer.length < requested) { buffer = new byte[requested]; } bufferPosition = 0; @@ -838,63 +781,6 @@ private boolean fillBuffer() throws IOException { return true; } - private void readCachedBlocks(Block first) throws IOException { - byte[] cached = cachedBlock(first); - if (cached != null) { - blockPosition++; - buffer = cached; - } else { - int firstPosition = blockPosition++; - long end = first.offset + first.length; - while (blockPosition < selected.blocks.size()) { - Block next = selected.blocks.get(blockPosition); - if (next.offset != end - || next.length > cache.maxElementSize() - || end - first.offset + next.length > BLOCK_READ_BUFFER_BYTES - || cachedBlock(next) != null) { - break; - } - end += next.length; - blockPosition++; - } - - byte[] bytes = new byte[(int) (end - first.offset)]; - seekInput(first.offset); - readFully(bytes, bytes.length); - // Publish only complete reads, and use individual block keys so overlapping - // selections can share data even when their coalesced read spans differ. - int offset = 0; - for (int i = firstPosition; i < blockPosition; i++) { - Block block = selected.blocks.get(i); - int length = (int) block.length; - byte[] blockBytes = - length == bytes.length - ? bytes - : Arrays.copyOfRange(bytes, offset, offset + length); - cache.put( - new BlockCacheKey(path, block.offset, block.length), - new ManifestBlockSegment(blockBytes)); - offset += length; - } - buffer = bytes; - } - bufferPosition = 0; - bufferLimit = buffer.length; - } - - @Nullable - private byte[] cachedBlock(Block block) { - Segments cached = - cache.getIfPresents(new BlockCacheKey(path, block.offset, block.length)); - if (cached instanceof ManifestBlockSegment) { - byte[] bytes = ((ManifestBlockSegment) cached).bytes; - if (bytes.length == block.length) { - return bytes; - } - } - return null; - } - private void seekInput(long offset) throws IOException { if (input == null) { input = io.newInputStream(path); 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 59106c5a2a70..4530f698e291 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,12 +63,6 @@ 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/main/java/org/apache/paimon/utils/ObjectsFile.java b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java index e6c923ef7c32..f8bec49019d3 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java @@ -167,7 +167,7 @@ public List read( return read(fileName, fileSize, readFilter, readTFilter, Function.identity()); } - private List readWithIOException( + protected List readWithIOException( String fileName, @Nullable Long fileSize, Filter readFilter, diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntryCacheTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntryCacheTest.java new file mode 100644 index 000000000000..4e558463b03b --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntryCacheTest.java @@ -0,0 +1,466 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.manifest; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.Segments; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.SeekableInputStreamWrapper; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.operation.metrics.CacheMetrics; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.schema.FileSystemSchemaManager; +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.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.paimon.TestKeyValueGenerator.DEFAULT_PART_TYPE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Cross-path correctness, isolation and bounded-memory tests for decoded manifest blocks. */ +class ManifestEntryCacheTest { + + @TempDir java.nio.file.Path temp; + + @Test + void partialFullAndCheckedReadsReuseTheSameEntries() throws Exception { + Fixture f = new Fixture(4000); + ManifestSidecar.Selection partial = f.selection(1, 3); + CacheMetrics metrics = new CacheMetrics(); + ManifestFile reader = f.reader(f.cache).withCacheMetrics(metrics); + assertThat(f.read(reader, partial)).containsExactlyElementsOf(f.expected(partial)); + assertThat(f.cache.getIfPresents(f.path)).isNull(); + ManifestSidecar.Block block = partial.blocks().get(0); + Segments pinned = f.cached(block); + assertThat(pinned).isInstanceOf(ManifestEntrySegments.class); + + // Without a complete directory, the first full read discovers the physical blocks. + assertThat(reader.readWithIOException(f.meta.fileName(), f.meta.fileSize())) + .containsExactlyElementsOf(f.entries); + assertThat(metrics.getHitObject().get()).isGreaterThanOrEqualTo(partial.blocks().size()); + assertThat(f.cached(block)).isSameAs(pinned); + assertThat(f.cache.getIfPresents(f.path)) + .isNotNull() + .isNotInstanceOf(ManifestEntrySegments.class); + + f.io.reset(); + f.io.rejectReads = true; + assertThat(f.read(f.reader(f.cache), partial)) + .containsExactlyElementsOf(f.expected(partial)); + assertThat(f.reader(f.cache).read(f.meta.fileName())).containsExactlyElementsOf(f.entries); + assertThat(f.reader(f.cache).readWithIOException(f.meta.fileName())) + .containsExactlyElementsOf(f.entries); + assertThat(f.io.opened).isEmpty(); + } + + @Test + void fullReadWarmsSubsequentSelectedRead() { + Fixture f = new Fixture(2000); + assertThat(f.reader(f.cache).read(f.meta.fileName())).containsExactlyElementsOf(f.entries); + f.io.reset(); + f.io.rejectReads = true; + ManifestSidecar.Selection selected = f.selection(0, 2); + assertThat(f.read(f.reader(f.cache), selected)) + .containsExactlyElementsOf(f.expected(selected)); + assertThat(f.io.opened).isEmpty(); + } + + @Test + void differentPredicatesAndConvertersNeverCachePartialEntries() { + Fixture f = new Fixture(1500); + List partitions = + f.entries.stream() + .map(ManifestEntry::partition) + .distinct() + .collect(Collectors.toList()); + assertThat(partitions.size()).isGreaterThan(1); + ManifestFile reader = f.reader(f.cache); + for (int i = 0; i < 2; i++) { + PartitionPredicate partition = + PartitionPredicate.fromMultiple( + DEFAULT_PART_TYPE, Collections.singletonList(partitions.get(i))); + int bucket = + f.entries.stream() + .filter(e -> partition.test(e.partition())) + .findFirst() + .get() + .bucket(); + BucketFilter buckets = new BucketFilter(false, bucket, null, null); + FileKind kind = i == 0 ? FileKind.ADD : FileKind.DELETE; + Filter predicate = e -> e.kind() == kind; + Function convertor = + i == 0 ? ManifestEntry::copyWithoutStats : Function.identity(); + List expected = + f.entries.stream() + .filter(e -> partition.test(e.partition())) + .filter(e -> buckets.test(e.partition(), e.bucket(), e.totalBuckets())) + .filter(predicate::test) + .map(convertor) + .collect(Collectors.toList()); + assertThat( + reader.read( + f.meta.fileName(), + f.meta.fileSize(), + partition, + buckets, + row -> true, + predicate, + convertor, + f.all)) + .containsExactlyElementsOf(expected); + f.io.reset(); + f.io.rejectReads = true; + } + assertThat(reader.read(f.meta.fileName())).containsExactlyElementsOf(f.entries); + assertThat(f.io.opened).isEmpty(); + } + + @Test + void mixedHitsAndMissesReadOnlyMissingRangesAndCoalesceNeighbours() { + Fixture f = new Fixture(4000); + ManifestFile reader = f.reader(f.cache); + f.read(reader, f.selection(1)); + f.io.reset(); + ManifestSidecar.Selection mixed = f.selection(0, 1, 2); + assertThat(f.read(reader, mixed)).containsExactlyElementsOf(f.expected(mixed)); + assertThat(f.io.seeks) + .containsExactly(f.all.blocks().get(0).offset, f.all.blocks().get(2).offset); + assertThat(f.io.bytes.get()) + .isEqualTo(f.all.blocks().get(0).length + f.all.blocks().get(2).length); + assertThat(f.io.opened).containsExactly(f.path); + f.io.reset(); + ManifestSidecar.Selection adjacent = f.selection(3, 4); + assertThat(f.read(reader, adjacent)).containsExactlyElementsOf(f.expected(adjacent)); + assertThat(f.io.seeks).containsExactly(f.all.blocks().get(3).offset); + assertThat(f.io.bytes.get()) + .isEqualTo(f.all.blocks().get(3).length + f.all.blocks().get(4).length); + } + + @Test + void truncatedBlockNeverPublishesDecodedEntriesOrCompleteDirectory() throws Exception { + Fixture f = new Fixture(1000); + ManifestSidecar.Selection selected = f.selection(1); + ManifestSidecar.Block block = selected.blocks().get(0); + java.nio.file.Path file = java.nio.file.Paths.get(f.path.toUri().getPath()); + byte[] original = Files.readAllBytes(file); + Files.write(file, Arrays.copyOf(original, (int) (block.offset + block.length - 1))); + f.io.reset(); + assertThatThrownBy(() -> f.read(f.reader(f.cache), selected)) + .hasRootCauseInstanceOf(EOFException.class); + assertThat(f.cached(block)).isNull(); + assertThat(f.cache.getIfPresents(f.path)).isNull(); + assertThat(f.io.closed.get()).isEqualTo(f.io.opened.size()); + Files.write(file, original); + assertThat(f.read(f.reader(f.cache), selected)) + .containsExactlyElementsOf(f.expected(selected)); + assertThat(f.cached(block)).isInstanceOf(ManifestEntrySegments.class); + } + + @Test + void decodedMemoryLimitFallsBackWithoutCachingAPrefixOrReadingTwice() { + Fixture f = new Fixture(1500); + SegmentsCache small = cache(8192, Long.MAX_VALUE); + ManifestFile reader = f.reader(small); + ManifestSidecar.Selection selected = f.selection(0, 1); + long bytes = selected.blocks().stream().mapToLong(b -> b.length).sum(); + for (int round = 0; round < 2; round++) { + f.io.reset(); + assertThat(f.read(reader, selected)).containsExactlyElementsOf(f.expected(selected)); + assertThat(f.io.bytes.get()).isEqualTo(bytes); + for (ManifestSidecar.Block block : selected.blocks()) { + assertThat( + blockCache(small) + .getIfPresents( + new BlockKey(f.path, block.offset, block.length))) + .isNull(); + } + assertThat(small.totalCacheBytes()).isLessThanOrEqualTo(8192); + } + } + + @Test + void evictionStaysWithinBudgetAndReloadsOnlyTheMissingBlocks() { + Fixture f = new Fixture(4000); + ManifestSidecar.Selection selected = f.selection(0, 1, 2, 3, 4); + f.read(f.reader(f.cache), selected); + long largest = + selected.blocks().stream() + .mapToLong(b -> f.cached(b).totalMemorySize()) + .max() + .getAsLong(); + long budget = largest * 2 + 3000; + SegmentsCache small = cache(budget, Long.MAX_VALUE); + ManifestFile reader = f.reader(small); + assertThat(f.read(reader, selected)).containsExactlyElementsOf(f.expected(selected)); + assertThat(small.totalCacheBytes()).isLessThanOrEqualTo(budget); + List missing = + selected.blocks().stream() + .filter( + b -> + blockCache(small) + .getIfPresents( + new BlockKey( + f.path, b.offset, b.length)) + == null) + .collect(Collectors.toList()); + assertThat(missing).isNotEmpty().hasSizeLessThan(selected.blocks().size()); + f.io.reset(); + assertThat(f.read(reader, selected)).containsExactlyElementsOf(f.expected(selected)); + assertThat(f.io.bytes.get()).isEqualTo(missing.stream().mapToLong(b -> b.length).sum()); + assertThat(small.totalCacheBytes()).isLessThanOrEqualTo(budget); + } + + @Test + void concurrentQueriesHaveIndependentReadersAndCursors() throws Exception { + Fixture f = new Fixture(4000); + ManifestFile shared = f.reader(f.cache); + ExecutorService executor = Executors.newFixedThreadPool(6); + try { + List> futures = new ArrayList<>(); + for (int task = 0; task < 12; task++) { + final int block = task % 6; + futures.add( + executor.submit( + () -> { + ManifestSidecar.Selection selected = + f.selection(block, block + 1); + for (int round = 0; round < 10; round++) { + assertThat(f.read(shared, selected)) + .containsExactlyElementsOf(f.expected(selected)); + } + })); + } + for (Future future : futures) { + future.get(30, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + } + + @Test + void blockIdentityIncludesPathOffsetAndLength() { + Path a = new Path("file:///a/manifest-1"); + Path b = new Path("file:///b/manifest-1"); + BlockKey key = new BlockKey(a, 100, 200); + assertThat(key).isEqualTo(new BlockKey(a, 100, 200)); + assertThat(key.hashCode()).isEqualTo(new BlockKey(a, 100, 200).hashCode()); + assertThat(key) + .isNotEqualTo(new BlockKey(b, 100, 200)) + .isNotEqualTo(new BlockKey(a, 101, 200)) + .isNotEqualTo(new BlockKey(a, 100, 201)); + } + + @SuppressWarnings("unchecked") + private static SegmentsCache blockCache(SegmentsCache cache) { + return (SegmentsCache) (SegmentsCache) cache; + } + + private static SegmentsCache cache(long memory, long element) { + return new SegmentsCache<>(1024, MemorySize.ofBytes(memory), element, null, false); + } + + private class Fixture { + + private final RecordingIO io = new RecordingIO(); + private final SegmentsCache cache = cache(64L << 20, 1L << 20); + private final SegmentsCache sidecars = cache(16L << 20, 1L << 20); + private final List entries = new ArrayList<>(); + private final Path root = new Path(temp.toString()); + private final ManifestFileMeta meta; + private final Path path; + private final ManifestSidecar.Selection all; + + private Fixture(int count) { + ManifestTestDataGenerator generator = ManifestTestDataGenerator.builder().build(); + for (int i = 0; i < count; i++) { + ManifestEntry e = generator.next(); + entries.add( + ManifestEntry.create( + i % 3 == 0 ? FileKind.DELETE : FileKind.ADD, + e.partition(), + e.bucket(), + e.totalBuckets(), + e.file().newFirstRowId(i * 1000000L))); + } + ManifestFile reader = reader(cache); + meta = reader.write(entries).get(0); + path = new Path(root, "manifest/" + meta.fileName()); + all = + reader.selectBlocks( + meta, + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + assertThat(all.blocks()).hasSizeGreaterThan(2); + io.reset(); + } + + private ManifestFile reader(SegmentsCache bodyCache) { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + FileStorePathFactory paths = + new FileStorePathFactory( + root, + DEFAULT_PART_TYPE, + "default", + CoreOptions.FILE_FORMAT.defaultValue(), + CoreOptions.DATA_FILE_PREFIX.defaultValue(), + CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(), + CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(), + CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(), + CoreOptions.FILE_COMPRESSION.defaultValue(), + null, + null, + CoreOptions.ExternalPathStrategy.NONE, + null, + false, + null); + return new ManifestFile.Factory( + io, + new FileSystemSchemaManager(io, root), + DEFAULT_PART_TYPE, + FileFormat.fromIdentifier("avro", new Options()), + "zstd", + paths, + Long.MAX_VALUE, + bodyCache, + sidecars, + new CoreOptions(options)) + .create(); + } + + private ManifestSidecar.Selection selection(int... indices) { + List blocks = new ArrayList<>(); + for (int index : indices) { + blocks.add(all.blocks().get(index)); + } + return new ManifestSidecar.Selection(all.header(), blocks); + } + + private List expected(ManifestSidecar.Selection selected) { + List result = new ArrayList<>(); + for (ManifestSidecar.Block b : selected.blocks()) { + result.addAll( + entries.subList( + (int) b.firstRecord, (int) (b.firstRecord + b.recordCount))); + } + return result; + } + + private List read(ManifestFile reader, ManifestSidecar.Selection selected) { + return reader.read( + meta.fileName(), + meta.fileSize(), + null, + null, + row -> true, + e -> true, + Function.identity(), + selected); + } + + private Segments cached(ManifestSidecar.Block block) { + return blockCache(cache).getIfPresents(new BlockKey(path, block.offset, block.length)); + } + } + + private static final class RecordingIO 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 final AtomicInteger closed = new AtomicInteger(); + private volatile boolean rejectReads; + + private void reset() { + opened.clear(); + seeks.clear(); + bytes.set(0); + closed.set(0); + } + + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + if (rejectReads) { + throw new IOException("Unexpected file read: " + path); + } + opened.add(path); + return new SeekableInputStreamWrapper(super.newInputStream(path)) { + + @Override + public int read() throws IOException { + int value = super.read(); + if (value >= 0) { + bytes.incrementAndGet(); + } + return value; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int count = super.read(b, off, len); + if (count > 0) { + bytes.addAndGet(count); + } + return count; + } + + @Override + public void seek(long offset) throws IOException { + seeks.add(offset); + super.seek(offset); + } + + @Override + public void close() throws IOException { + closed.incrementAndGet(); + super.close(); + } + }; + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySegmentsTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySegmentsTest.java index 4c1a2deb6cfa..6403fe130630 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySegmentsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySegmentsTest.java @@ -64,8 +64,13 @@ public void testManifestEntrySegments() { assertThat(manifestEntrySegments.segments()).containsExactlyElementsOf(richSegmentsList); // Test totalMemorySize() method - long expectedTotalMemorySize = 100 + 200 + 300; - assertThat(manifestEntrySegments.totalMemorySize()).isEqualTo(expectedTotalMemorySize); + long expectedTotalMemorySize = + richSegments1.totalMemorySize() + + richSegments2.totalMemorySize() + + richSegments3.totalMemorySize(); + assertThat(manifestEntrySegments.totalMemorySize()) + .isEqualTo(expectedTotalMemorySize) + .isGreaterThan(100 + 200 + 300); // Test indexedSegments() method Map>> indexedSegments = diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index db914239eea5..5d9c77ac933a 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 @@ -1471,7 +1471,7 @@ void testSidecarCacheUsesExplicitPathsAndIsSeparateFromManifestCache() throws Ex assertThat(io.opened).isEmpty(); assertThat(factory.create().read(meta.fileName())) .containsExactlyInAnyOrderElementsOf(entries); - assertThat(cache.estimatedSize()).isEqualTo(1); + assertThat(cache.estimatedSize()).isEqualTo(2); assertThat(sidecarCache.estimatedSize()).isEqualTo(1); assertThat(cache.getIfPresents(sidecarPath)).isNull(); assertThat( @@ -1554,8 +1554,9 @@ void testDedicatedSidecarCacheAndManifestCacheFallback() { .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); + // One decoded block plus its complete directory; sidecar bytes join only on + // fallback. + assertThat(manifestCache.estimatedSize()).isEqualTo(cacheSidecar ? 2 : 3); } if (sidecarCache != null) { assertThat(sidecarCache.estimatedSize()).isEqualTo(1); @@ -1631,17 +1632,21 @@ void testReadsOnlySelectedBlocksAndPreservesPhysicalOrdinals() throws Exception assertThat(readSelectedEntries(factory.create(), meta, allBlocks)) .containsExactlyInAnyOrderElementsOf(entries); assertThat(fileIO.opened).containsExactly(manifestPath); - assertThat(cache.getIfPresents(manifestPath)).isNull(); + assertThat(cache.getIfPresents(manifestPath)) + .isNotNull() + .isNotInstanceOf(ManifestEntrySegments.class); fileIO.reset(); assertThat(readSelectedEntries(factory.create(), meta, allBlocks)) .containsExactlyInAnyOrderElementsOf(entries); assertThat(fileIO.opened).isEmpty(); - assertThat(cache.getIfPresents(manifestPath)).isNull(); + assertThat(cache.getIfPresents(manifestPath)) + .isNotNull() + .isNotInstanceOf(ManifestEntrySegments.class); - // Reads without a sidecar selection populate and reuse the full-manifest cache. + // Full reads reuse exactly the same decoded blocks, not a second copy of all entries. assertThat(manifests.read(meta.fileName())).containsExactlyInAnyOrderElementsOf(entries); - assertThat(fileIO.opened).containsExactly(manifestPath); + assertThat(fileIO.opened).isEmpty(); assertThat(cache.getIfPresents(manifestPath)).isNotNull(); fileIO.reset(); @@ -1652,7 +1657,7 @@ void testReadsOnlySelectedBlocksAndPreservesPhysicalOrdinals() throws Exception 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); + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(16), 1 << 20, null, false); ManifestFile.Factory limitedFactory = createManifestFileFactory( tempDir.toString(), Long.MAX_VALUE, options, fileIO, blockCache); @@ -1667,7 +1672,8 @@ void testReadsOnlySelectedBlocksAndPreservesPhysicalOrdinals() throws Exception assertThat( blockCache.getIfPresents( new Path(tempDir.toString(), "manifest/" + meta.fileName()))) - .isNull(); + .isNotNull() + .isNotInstanceOf(ManifestEntrySegments.class); } @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 1b3f9d258ee4..be738948d753 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 @@ -811,203 +811,6 @@ io, path, select(testSidecar(), testMeta(), point))) { } } - @Test - void cachedBlocksAreSharedByDifferentSelectionsWithoutOpeningTheManifest() throws Exception { - byte[] header = header(); - byte[] manifest = Arrays.copyOf(header, header.length + 400); - for (int i = header.length; i < manifest.length; i++) { - manifest[i] = (byte) i; - } - Path path = new Path(temp.toString(), "manifest-golden"); - SegmentsCache cache = - new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), 400, null, false); - cache.put( - new ManifestSidecar.BlockCacheKey(path, header.length, 100), - new SingleSegments(MemorySegment.wrap(new byte[100]), 100)); - FileIO io = mock(FileIO.class); - CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); - when(io.newInputStream(path)).thenReturn(stream); - ManifestSidecar.Selection all = - ManifestSidecar.select( - testSidecar(), - testMeta(), - RowRangeIndex.create( - Collections.singletonList(new Range(0, Long.MAX_VALUE)))); - try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); - } - assertThat(stream.readLengths).containsExactly(400); - assertThat(stream.seeks).containsExactly((long) header.length); - assertThat(cache.estimatedSize()).isEqualTo(3); - assertThat(cache.getIfPresents(path)).isNull(); - when(io.newInputStream(path)).thenThrow(new IOException("Must use cached blocks")); - for (long point : new long[] {20, 8254058425445L, 0}) { - ManifestSidecar.Selection selected = select(testSidecar(), testMeta(), point); - ByteArrayOutputStream expected = new ByteArrayOutputStream(); - expected.write(header); - for (ManifestSidecar.Block block : selected.blocks()) { - expected.write(manifest, (int) block.offset, (int) block.length); - } - try (InputStream in = ManifestSidecar.openManifest(io, path, selected, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(expected.toByteArray()); - } - } - verify(io, times(1)).newInputStream(path); - - Path other = new Path(temp.toString(), "other/manifest-golden"); - byte[] otherBytes = manifest.clone(); - otherBytes[header.length] ^= 1; - when(io.newInputStream(other)).thenReturn(new CountingInput(otherBytes, Integer.MAX_VALUE)); - try (InputStream in = - ManifestSidecar.openManifest( - io, other, select(testSidecar(), testMeta(), 0), cache)) { - assertThat(IOUtils.readFully(in, false)) - .isEqualTo(Arrays.copyOf(otherBytes, header.length + 100)); - } - verify(io).newInputStream(other); - } - - @Test - void mixedHitsAndMissesReadOnlyUncachedBlocks() throws Exception { - byte[] header = header(); - byte[] manifest = Arrays.copyOf(header, header.length + 400); - FileIO io = mock(FileIO.class); - Path path = new Path(temp.toString(), "manifest-golden"); - CountingInput cold = new CountingInput(manifest, Integer.MAX_VALUE); - CountingInput mixed = new CountingInput(manifest, Integer.MAX_VALUE); - when(io.newInputStream(path)).thenReturn(cold, mixed); - SegmentsCache cache = - new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), 400, null, false); - try (InputStream in = - ManifestSidecar.openManifest( - io, path, select(testSidecar(), testMeta(), 8254058425445L), cache)) { - IOUtils.readFully(in, false); - } - ManifestSidecar.Selection all = - ManifestSidecar.select( - testSidecar(), - testMeta(), - RowRangeIndex.create( - Collections.singletonList(new Range(0, Long.MAX_VALUE)))); - try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); - } - assertThat(mixed.readLengths).containsExactly(100, 100); - assertThat(mixed.seeks).containsExactly((long) header.length, header.length + 300L); - try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); - } - verify(io, times(2)).newInputStream(path); - } - - @Test - void truncatedCoalescedReadsDoNotPopulateTheBlockCache() throws Exception { - byte[] header = header(); - byte[] manifest = Arrays.copyOf(header, header.length + 400); - FileIO io = mock(FileIO.class); - Path path = new Path(temp.toString(), "manifest-golden"); - CountingInput truncated = - new CountingInput(Arrays.copyOf(manifest, manifest.length - 1), 7); - CountingInput complete = new CountingInput(manifest, 7); - when(io.newInputStream(path)).thenReturn(truncated, complete); - SegmentsCache cache = - new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), 400, null, false); - ManifestSidecar.Selection all = - ManifestSidecar.select( - testSidecar(), - testMeta(), - RowRangeIndex.create( - Collections.singletonList(new Range(0, Long.MAX_VALUE)))); - try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { - assertThatThrownBy(() -> IOUtils.readFully(in, false)).isInstanceOf(EOFException.class); - } - assertThat(truncated.closed).isTrue(); - assertThat(cache.estimatedSize()).isZero(); - try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); - } - assertThat(complete.closed).isTrue(); - assertThat(cache.estimatedSize()).isEqualTo(3); - } - - @Test - void evictedBlocksAreReadAgainWithinTheSharedBudget() throws Exception { - byte[] header = header(); - byte[] manifest = Arrays.copyOf(header, header.length + 400); - Path path = new Path(temp.toString(), "manifest-golden"); - FileIO io = mock(FileIO.class); - when(io.newInputStream(path)) - .thenAnswer(ignored -> new CountingInput(manifest, Integer.MAX_VALUE)); - SegmentsCache cache = - new SegmentsCache<>(1024, MemorySize.ofBytes(1300), 400, null, false); - ManifestSidecar.Selection all = - ManifestSidecar.select( - testSidecar(), - testMeta(), - RowRangeIndex.create( - Collections.singletonList(new Range(0, Long.MAX_VALUE)))); - for (int round = 0; round < 2; round++) { - try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); - } - assertThat(cache.totalCacheBytes()).isLessThanOrEqualTo(1300); - assertThat(cache.estimatedSize()).isEqualTo(1); - } - verify(io, times(2)).newInputStream(path); - } - - @Test - void oversizedBlocksUseBoundedReadsWithoutModifyingPreviouslyCachedBytes() throws Exception { - byte[] header = header(); - int cachedLength = (4 << 20) + 17; - int uncachedLength = 2 * (4 << 20) + 31; - ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); - builder.beginBlock(header.length, cachedLength, 1); - builder.add(0L, 1); - builder.endBlock(); - builder.beginBlock(header.length + cachedLength, uncachedLength, 1); - builder.add(100L, 1); - builder.endBlock(); - byte[] manifest = Arrays.copyOf(header, header.length + cachedLength + uncachedLength); - Arrays.fill(manifest, header.length, header.length + cachedLength, (byte) 7); - Arrays.fill(manifest, header.length + cachedLength, manifest.length, (byte) 9); - byte[] data = builder.serialize(manifest.length, 2); - ManifestFileMeta meta = meta("large", manifest.length, 2); - Path path = new Path(temp.toString(), "large"); - FileIO io = mock(FileIO.class); - CountingInput cold = new CountingInput(manifest, Integer.MAX_VALUE); - CountingInput mixed = new CountingInput(manifest, Integer.MAX_VALUE); - when(io.newInputStream(path)).thenReturn(cold, mixed); - SegmentsCache cache = - new SegmentsCache<>(1024, MemorySize.ofMebiBytes(8), cachedLength, null, false); - ManifestSidecar.Selection first = select(data, meta, 0); - byte[] expected = Arrays.copyOf(manifest, header.length + cachedLength); - try (InputStream in = ManifestSidecar.openManifest(io, path, first, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(expected); - } - assertThat(cold.readLengths).containsExactly(4 << 20, 17); - ManifestSidecar.Selection all = - ManifestSidecar.select( - data, - meta, - RowRangeIndex.create( - Collections.singletonList(new Range(0, Long.MAX_VALUE)))); - try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); - } - assertThat(mixed.readLengths).containsExactly(4 << 20, 4 << 20, 31); - assertThat(cache.estimatedSize()).isEqualTo(1); - assertThat( - cache.getIfPresents( - new ManifestSidecar.BlockCacheKey( - path, header.length + cachedLength, uncachedLength))) - .isNull(); - try (InputStream in = ManifestSidecar.openManifest(io, path, first, cache)) { - assertThat(IOUtils.readFully(in, false)).isEqualTo(expected); - } - verify(io, times(2)).newInputStream(path); - } - @Test void largeBlockSpansUseBoundedReads() throws Exception { byte[] header = header(); @@ -1022,33 +825,17 @@ void largeBlockSpansUseBoundedReads() throws Exception { byte[] data = builder.serialize(offset, 5); byte[] manifest = Arrays.copyOf(header, (int) offset); Path path = new Path(temp.toString(), "manifest-large"); - for (boolean withCache : new boolean[] {false, true}) { - CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); - FileIO io = mock(FileIO.class); - when(io.newInputStream(path)).thenReturn(stream); - SegmentsCache cache = - withCache - ? new SegmentsCache<>( - 1024, MemorySize.ofMebiBytes(16), 4 << 20, null, false) - : null; - try (InputStream input = - ManifestSidecar.openManifest( - io, path, select(data, meta("manifest-large", offset, 5), 20), cache)) { - assertThat(IOUtils.readFully(input, false)).isEqualTo(manifest); - } - assertThat(stream.readLengths).containsExactly(4 << 20, 4 << 20, 1 << 20); - if (withCache) { - assertThat(stream.seeks) - .containsExactly( - (long) header.length, - header.length + (4L << 20), - header.length + (8L << 20)); - assertThat(cache.estimatedSize()).isEqualTo(5); - } else { - assertThat(stream.seeks).containsExactly((long) header.length); - } - assertThat(stream.closed).isTrue(); + CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); + FileIO io = mock(FileIO.class); + when(io.newInputStream(path)).thenReturn(stream); + try (InputStream input = + ManifestSidecar.openManifest( + io, path, select(data, meta("manifest-large", offset, 5), 20))) { + assertThat(IOUtils.readFully(input, false)).isEqualTo(manifest); } + assertThat(stream.readLengths).containsExactly(4 << 20, 4 << 20, 1 << 20); + assertThat(stream.seeks).containsExactly((long) header.length); + assertThat(stream.closed).isTrue(); } @Test