Skip to content

Commit 084cde4

Browse files
leaves12138XiaoHongbo-Hope
authored andcommitted
[core] Prune manifest blocks with sidecars during scans (#9897)
1 parent ce81e89 commit 084cde4

23 files changed

Lines changed: 1022 additions & 37 deletions

File tree

‎docs/docs/concepts/spec/manifest.md‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -72,20 +72,28 @@ using independent partition, row-ID and bucket coverage. A sidecar uses the
7272
derived file name. The Avro schemas and `_VERSION` identifiers remain unchanged.
7373

7474
The utility includes construction, validation, block selection and optional caching. Java table
75-
writers generate sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits
75+
writers and scans use sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits
7676
`manifest-sort.enabled`. Both ordinary writes and raw manifest rewrites build the sidecar from
7777
the completed output manifest and publish its `_EXTRA_FILES` reference only after both files
7878
close successfully. Failed writes and aborted writers clean up their own manifest/sidecar pairs.
79-
Scans do not yet invoke sidecar pruning automatically. Callers remain responsible for applying
80-
entry filters and reconciling ADD/DELETE entries after block selection. The low-level `build`
81-
method returns sidecar bytes without writing or publishing another file.
79+
Scans with partition, row-ID or bucket filters select blocks before reading manifest entries.
80+
Normal entry filtering and ADD/DELETE reconciliation still apply. Missing or unusable sidecars
81+
fall back to normal manifest reads; disabled sidecars and scans without these filters do not
82+
perform sidecar I/O. Sidecar caching is controlled by the catalog option
83+
`cache.manifest-sidecar.max-memory` (64 MiB by default). A positive value supplies an
84+
additional budget independent of the manifest content cache. When set to 0, sidecars reuse
85+
the manifest content cache, or remain uncached if that cache is disabled. Sidecar caching
86+
uses the catalog's `cache.expire-after-access` and `cache.manifest.soft-values` policies.
87+
Selected block bytes still share the manifest content cache without populating the
88+
whole-manifest entry cache with partial results. The low-level `build` method returns
89+
sidecar bytes without writing or publishing another file.
8290

8391
Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches.
8492
`build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent
8593
payload generation. Partition generation is always enabled,
8694
including the empty partition tuple for unpartitioned tables. Missing or invalid
87-
metadata makes only the affected block's dimension unavailable. There is no sidecar byte budget:
88-
construction keeps complete coverage and `read` consumes the entire file once it is opened.
95+
metadata makes only the affected block's dimension unavailable. Cache limits do not truncate
96+
sidecars: construction keeps complete coverage and `read` consumes the entire file once it is opened.
8997

9098
`read` returns null for an absent sidecar reference or an `IOException`, allowing the caller
9199
to fall back to the manifest. If the thread is interrupted, the I/O failure is propagated as

‎docs/generated/catalog_configuration.html‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,12 @@
5050
<td>Duration</td>
5151
<td>Cache expiration policy: marks cache entries to expire after a specified duration has passed since their last refresh.</td>
5252
</tr>
53+
<tr>
54+
<td><h5>cache.manifest-sidecar.max-memory</h5></td>
55+
<td style="word-wrap: break-word;">64 mb</td>
56+
<td>MemorySize</td>
57+
<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>
58+
</tr>
5359
<tr>
5460
<td><h5>cache.manifest.max-memory</h5></td>
5561
<td style="word-wrap: break-word;">(none)</td>

‎paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,6 +130,16 @@ public class CatalogOptions {
130130
.noDefaultValue()
131131
.withDescription("Controls the maximum memory to cache manifest content.");
132132

133+
public static final ConfigOption<MemorySize> CACHE_MANIFEST_SIDECAR_MAX_MEMORY =
134+
key("cache.manifest-sidecar.max-memory")
135+
.memoryType()
136+
.defaultValue(MemorySize.ofMebiBytes(64))
137+
.withDescription(
138+
"Controls the maximum memory for the separate manifest sidecar cache. "
139+
+ "This budget is additional to the manifest content cache. "
140+
+ "Set to 0 to reuse the manifest content cache. It uses "
141+
+ "'cache.expire-after-access' and 'cache.manifest.soft-values'.");
142+
133143
public static final ConfigOption<Boolean> CACHE_MANIFEST_SOFT_VALUES =
134144
key("cache.manifest.soft-values")
135145
.booleanType()

‎paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -301,6 +301,7 @@ private ManifestFile createManifestFile() {
301301
pathFactory,
302302
TARGET_MANIFEST_SIZE,
303303
null,
304+
null,
304305
new CoreOptions(new Options()))
305306
.create();
306307
}

‎paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@ abstract class AbstractFileStore<T> implements FileStore<T> {
109109
protected final CatalogEnvironment catalogEnvironment;
110110

111111
@Nullable private SegmentsCache<Path> readManifestCache;
112+
@Nullable private SegmentsCache<Path> manifestSidecarCache;
112113
@Nullable private Cache<Path, Snapshot> snapshotCache;
113114

114115
protected AbstractFileStore(
@@ -212,6 +213,7 @@ public ManifestFile.Factory manifestFileFactory() {
212213
pathFactory(),
213214
options.manifestTargetSize().getBytes(),
214215
readManifestCache,
216+
manifestSidecarCache,
215217
options);
216218
}
217219

@@ -616,6 +618,11 @@ public void setManifestCache(SegmentsCache<Path> manifestCache) {
616618
this.readManifestCache = manifestCache;
617619
}
618620

621+
@Override
622+
public void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) {
623+
this.manifestSidecarCache = manifestSidecarCache;
624+
}
625+
619626
@Override
620627
public void setSnapshotCache(Cache<Path, Snapshot> cache) {
621628
this.snapshotCache = cache;

‎paimon-core/src/main/java/org/apache/paimon/FileStore.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,5 +127,7 @@ boolean mergeSchema(
127127

128128
void setManifestCache(SegmentsCache<Path> manifestCache);
129129

130+
void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache);
131+
130132
void setSnapshotCache(Cache<Path, Snapshot> cache);
131133
}

‎paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
import static org.apache.paimon.options.CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS;
5252
import static org.apache.paimon.options.CatalogOptions.CACHE_EXPIRE_AFTER_WRITE;
5353
import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_MAX_MEMORY;
54+
import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SIDECAR_MAX_MEMORY;
5455
import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_MEMORY;
5556
import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_THRESHOLD;
5657
import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SOFT_VALUES;
@@ -71,6 +72,7 @@ public class CachingCatalog extends DelegateCatalog {
7172
protected Cache<String, Database> databaseCache;
7273
protected Cache<Identifier, Table> tableCache;
7374
@Nullable protected final SegmentsCache<Path> manifestCache;
75+
@Nullable protected final SegmentsCache<Path> manifestSidecarCache;
7476
// partition cache will affect data latency
7577
@Nullable protected Cache<Identifier, List<Partition>> partitionCache;
7678
@Nullable protected DVMetaCache dvMetaCache;
@@ -108,6 +110,17 @@ public CachingCatalog(Catalog wrapped, Options options) {
108110
expireAfterAccess,
109111
manifestCacheSoftValues);
110112

113+
MemorySize sidecarMaxMemory = options.get(CACHE_MANIFEST_SIDECAR_MAX_MEMORY);
114+
this.manifestSidecarCache =
115+
sidecarMaxMemory.getBytes() == 0
116+
? manifestCache
117+
: SegmentsCache.create(
118+
(int) CoreOptions.PAGE_SIZE.defaultValue().getBytes(),
119+
sidecarMaxMemory,
120+
sidecarMaxMemory.getBytes(),
121+
expireAfterAccess,
122+
manifestCacheSoftValues);
123+
111124
this.cachedPartitionMaxNum = options.get(CACHE_PARTITION_MAX_NUM);
112125

113126
int cacheDvMaxNum = options.get(CACHE_DV_MAX_NUM);
@@ -299,6 +312,9 @@ private Table loadTable(Identifier identifier) {
299312
if (manifestCache != null) {
300313
storeTable.setManifestCache(manifestCache);
301314
}
315+
if (manifestSidecarCache != null) {
316+
storeTable.setManifestSidecarCache(manifestSidecarCache);
317+
}
302318
if (dvMetaCache != null) {
303319
storeTable.setDVMetaCache(dvMetaCache);
304320
}
@@ -425,6 +441,11 @@ public CacheSizes estimatedCacheSizes() {
425441
manifestCacheSize = manifestCache.estimatedSize();
426442
manifestCacheBytes = manifestCache.totalCacheBytes();
427443
}
444+
// Keep reporting total manifest metadata usage even though sidecars have their own budget.
445+
if (manifestSidecarCache != null && manifestSidecarCache != manifestCache) {
446+
manifestCacheSize += manifestSidecarCache.estimatedSize();
447+
manifestCacheBytes += manifestSidecarCache.totalCacheBytes();
448+
}
428449
long partitionCacheSize = 0L;
429450
if (partitionCache != null) {
430451
for (Map.Entry<Identifier, List<Partition>> entry : partitionCache.asMap().entrySet()) {

‎paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,22 @@ public boolean test(BinaryRow partition, int bucket, int totalBucket) {
7878
|| totalAwareBucketFilter.test(partition, bucket, totalBucket);
7979
}
8080

81+
/** Conservatively checks an indexed pair without inventing a partition for custom filters. */
82+
public boolean mayContain(int bucket, int totalBuckets) {
83+
if (onlyReadRealBuckets && bucket < 0) {
84+
return false;
85+
}
86+
if (specifiedBucket != null && bucket != specifiedBucket) {
87+
return false;
88+
}
89+
if (bucketFilter != null && !bucketFilter.test(bucket)) {
90+
return false;
91+
}
92+
return !(totalAwareBucketFilter instanceof ManifestBucketFilter)
93+
|| ((ManifestBucketFilter) totalAwareBucketFilter)
94+
.mayContain(bucket, bucket, totalBuckets);
95+
}
96+
8197
/** Conservatively tests whether a manifest's bucket metadata can contain a matching entry. */
8298
public boolean mayContain(ManifestFileMeta manifest) {
8399
Integer minBucket = manifest.minBucket();

‎paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java‎

Lines changed: 78 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.apache.paimon.utils.Filter;
3838
import org.apache.paimon.utils.ObjectsFile;
3939
import org.apache.paimon.utils.PathFactory;
40+
import org.apache.paimon.utils.RowRangeIndex;
4041
import org.apache.paimon.utils.SegmentsCache;
4142

4243
import javax.annotation.Nullable;
@@ -61,6 +62,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> {
6162
private final AvroFileFormat avroFileFormat;
6263
private final long suggestedFileSize;
6364
private final CoreOptions options;
65+
@Nullable private final SegmentsCache<Path> sidecarCache;
6466

6567
private ManifestFile(
6668
FileIO fileIO,
@@ -72,6 +74,7 @@ private ManifestFile(
7274
PathFactory pathFactory,
7375
long suggestedFileSize,
7476
@Nullable SegmentsCache<Path> cache,
77+
@Nullable SegmentsCache<Path> sidecarCache,
7578
CoreOptions options) {
7679
super(
7780
fileIO,
@@ -89,6 +92,7 @@ private ManifestFile(
8992
this.avroFileFormat = avroFileFormat;
9093
this.suggestedFileSize = suggestedFileSize;
9194
this.options = options;
95+
this.sidecarCache = sidecarCache == null ? cache : sidecarCache;
9296
}
9397

9498
@Override
@@ -140,9 +144,33 @@ public <T> List<T> read(
140144
Filter<InternalRow> readFilter,
141145
Filter<ManifestEntry> readTFilter,
142146
Function<ManifestEntry, T> convertor) {
147+
return read(
148+
fileName,
149+
fileSize,
150+
partitionFilter,
151+
bucketFilter,
152+
readFilter,
153+
readTFilter,
154+
convertor,
155+
null);
156+
}
157+
158+
public <T> List<T> read(
159+
String fileName,
160+
@Nullable Long fileSize,
161+
@Nullable PartitionPredicate partitionFilter,
162+
@Nullable BucketFilter bucketFilter,
163+
Filter<InternalRow> readFilter,
164+
Filter<ManifestEntry> readTFilter,
165+
Function<ManifestEntry, T> convertor,
166+
@Nullable ManifestSidecar.Selection selected) {
167+
if (selected != null && selected.blocks().isEmpty()) {
168+
return java.util.Collections.emptyList();
169+
}
143170
try {
144171
Path path = pathFactory.toPath(fileName);
145-
if (cache != null) {
172+
// Sidecar selections use the block cache, even when every block is selected.
173+
if (cache != null && selected == null) {
146174
ManifestEntryFilters filters =
147175
new ManifestEntryFilters(
148176
partitionFilter, bucketFilter, readFilter, readTFilter);
@@ -155,7 +183,9 @@ public <T> List<T> read(
155183
path,
156184
ManifestEntry.MANIFEST_ROW_TYPE,
157185
partitionFilter,
158-
bucketFilter);
186+
bucketFilter,
187+
selected,
188+
cache == null ? null : cache.segmentsCache());
159189
return readFromIterator(iterator, serializer, readFilter, readTFilter, convertor);
160190
} catch (IOException e) {
161191
throw new UncheckedIOException(e);
@@ -209,8 +239,23 @@ private static CloseableIterator<InternalRow> createManifestIterator(
209239
@Nullable PartitionPredicate partitionFilter,
210240
@Nullable BucketFilter bucketFilter)
211241
throws IOException {
242+
return createManifestIterator(
243+
fileIO, path, projectedType, partitionFilter, bucketFilter, null, null);
244+
}
245+
246+
private static CloseableIterator<InternalRow> createManifestIterator(
247+
FileIO fileIO,
248+
Path path,
249+
RowType projectedType,
250+
@Nullable PartitionPredicate partitionFilter,
251+
@Nullable BucketFilter bucketFilter,
252+
@Nullable ManifestSidecar.Selection selected,
253+
@Nullable SegmentsCache<Object> cache)
254+
throws IOException {
212255
try {
213-
ManifestAvroReader reader = new ManifestAvroReader(fileIO.newInputStream(path));
256+
ManifestAvroReader reader =
257+
new ManifestAvroReader(
258+
ManifestSidecar.openManifest(fileIO, path, selected, cache));
214259
return reader.read(projectedType, partitionFilter, bucketFilter);
215260
} catch (IOException e) {
216261
FileUtils.checkExists(fileIO, path);
@@ -345,6 +390,32 @@ public Path toPath(String fileName) {
345390
};
346391
}
347392

393+
@Nullable
394+
public ManifestSidecar.Selection selectBlocks(
395+
ManifestFileMeta manifest, @Nullable RowRangeIndex query) {
396+
return selectBlocks(manifest, query, null, null);
397+
}
398+
399+
@Nullable
400+
public ManifestSidecar.Selection selectBlocks(
401+
ManifestFileMeta manifest,
402+
@Nullable RowRangeIndex query,
403+
@Nullable PartitionPredicate partitionFilter,
404+
@Nullable BucketFilter bucketFilter) {
405+
return !options.manifestSidecarEnabled()
406+
|| (query == null && partitionFilter == null && bucketFilter == null)
407+
? null
408+
: ManifestSidecar.read(
409+
fileIO,
410+
pathFactory.toPath(manifest.fileName()),
411+
manifest,
412+
query,
413+
partitionFilter,
414+
partitionType,
415+
bucketFilter == null ? null : bucketFilter::mayContain,
416+
sidecarCache);
417+
}
418+
348419
/** Deletes an unreferenced manifest and its explicitly referenced extra files. */
349420
public void delete(ManifestFileMeta manifest) {
350421
delete(manifest.fileName());
@@ -365,6 +436,7 @@ public static class Factory {
365436
private final long suggestedFileSize;
366437
private final CoreOptions options;
367438
@Nullable private final SegmentsCache<Path> cache;
439+
@Nullable private final SegmentsCache<Path> sidecarCache;
368440

369441
public Factory(
370442
FileIO fileIO,
@@ -375,6 +447,7 @@ public Factory(
375447
FileStorePathFactory pathFactory,
376448
long suggestedFileSize,
377449
@Nullable SegmentsCache<Path> cache,
450+
@Nullable SegmentsCache<Path> sidecarCache,
378451
CoreOptions options) {
379452
this.fileIO = fileIO;
380453
this.schemaManager = schemaManager;
@@ -384,13 +457,10 @@ public Factory(
384457
this.pathFactory = pathFactory;
385458
this.suggestedFileSize = suggestedFileSize;
386459
this.cache = cache;
460+
this.sidecarCache = sidecarCache;
387461
this.options = options;
388462
}
389463

390-
public boolean isCacheEnabled() {
391-
return cache != null;
392-
}
393-
394464
public ManifestFile create() {
395465
return new ManifestFile(
396466
fileIO,
@@ -402,6 +472,7 @@ public ManifestFile create() {
402472
pathFactory.manifestFileFactory(),
403473
suggestedFileSize,
404474
cache,
475+
sidecarCache,
405476
options);
406477
}
407478
}

‎paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -632,7 +632,7 @@ public static Selection read(
632632
@Nullable PartitionPredicate partitionFilter,
633633
@Nullable RowType partitionType,
634634
@Nullable BiPredicate<Integer, Integer> bucketFilter,
635-
@Nullable SegmentsCache<Object> cache) {
635+
@Nullable SegmentsCache<Path> cache) {
636636
String sidecarFileName = fileName(manifest);
637637
if (sidecarFileName == null) {
638638
return null;
@@ -660,7 +660,7 @@ public static Selection read(
660660
}
661661
}
662662

663-
/** Complete sidecar bytes stored in the shared manifest cache. */
663+
/** Complete sidecar bytes stored in the configured sidecar cache. */
664664
static final class ManifestSidecarSegment implements Segments {
665665
private final byte[] bytes;
666666

0 commit comments

Comments
 (0)