capistrant commented on code in PR #19671:
URL: https://github.com/apache/druid/pull/19671#discussion_r3554948425
##########
server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java:
##########
@@ -199,6 +230,326 @@ File getLocalCacheDir()
return localCacheDir;
}
+ /**
+ * The fingerprint of the partial-load rule currently applied to this entry,
or {@code null} if no rule has been
+ * applied. Set by {@link #applyRule(String, Set)} and cleared by {@link
#clearRule()}. {@link #doActualUnmount()}
+ * asserts that this field is {@code null} at unmount time, the {@code
metadataSelfHold} taken by applyRule pins
+ * the entry's phaser and prevents eviction reclaim / doActualUnmount from
firing while a rule is applied. A fresh
+ * mount therefore always starts with no rule applied.
+ */
+ @Nullable
+ public String getRuleFingerprint()
+ {
+ entryLock.lock();
+ try {
+ return ruleFingerprint;
+ }
+ finally {
+ entryLock.unlock();
+ }
+ }
+
+ /**
+ * The set of bundle names currently pinned by an applied partial-load rule,
or an empty set if no rule is applied.
+ * Return value is immutable; safe to inspect without further
synchronization.
+ */
+ public Set<String> getRuleSelectedBundleNames()
+ {
+ entryLock.lock();
+ try {
+ return ruleSelectedBundleNames;
+ }
+ finally {
+ entryLock.unlock();
+ }
+ }
+
+ /**
+ * Whether a partial-load rule is currently applied to this entry.
Equivalent to {@code getRuleFingerprint() != null}.
+ * While {@code true}, this entry holds a self-referential {@link
StorageLocation.ReservationHold} that prevents cache
+ * from evicting it, so the metadata's V10 header stays resident until
{@link #clearRule()} runs.
+ */
+ public boolean isRuleHeld()
+ {
+ entryLock.lock();
+ try {
+ return ruleFingerprint != null;
+ }
+ finally {
+ entryLock.unlock();
+ }
+ }
+
+ /**
+ * Apply (or replace) a partial-load rule on this entry. Must be called
after {@link #mount(StorageLocation)}, the
+ * entry must be registered with its {@link StorageLocation} so it can
acquire holds on itself and on its bundles.
+ * <p>
+ * <b>Concurrency contract.</b> The caller MUST serialize {@code applyRule}
and {@link #clearRule} for a given
+ * segment id through an external per-segment lock ({@code
SegmentLocalCacheManager} uses its
+ * {@code ReferenceCountingLock segmentLocks}). This method releases {@link
#entryLock} between Phase 2 (acquire
+ * holds outside the lock) and Phase 4 (release holds outside the lock); a
concurrent invocation of
+ * {@code applyRule}/{@code clearRule} on the same entry can observe
intermediate state during those windows and
+ * silently miss rule-hold installations or leak them. {@link #entryLock}
alone is insufficient because it is
+ * released across the acquire/release phases by design (holding it across
{@link StorageLocation} lock acquisition
+ * would invert the writeLock → entryLock ordering the storage layer
already uses for eviction).
+ * <p>
+ * On first application ({@code ruleFingerprint} transitions from {@code
null}), a self-referential
+ * {@link StorageLocation.ReservationHold} is taken to pin the metadata
entry in-cache. On repeat calls with the same
+ * {@code fingerprint} and matching {@code selectedBundleNames}, this is a
no-op. On a genuine rule swap, only the
+ * delta between the previous and new selection is applied: bundle holds for
names dropped from the selection are
+ * closed, and holds for names newly added are acquired for any bundle
already registered with this entry (later
+ * arrivals get their hold via {@link #registerBundle}).
+ * <p>
+ * Bundles whose {@link StorageLocation.ReservationHold} is currently
zero-refcount (never mounted, or evicted but
+ * unregistered before this call) will not appear in {@link #linkedBundles}
yet, so no hold is taken for them here;
+ * the caller is expected to drive an on-demand acquire per selected bundle
name to trigger a fresh mount, which will
+ * call {@link #registerBundle} and pick up the rule-hold at that point.
+ *
+ * @throws DruidException if the entry is not currently mounted
+ */
+ public void applyRule(String fingerprint, Set<String> selectedBundleNames)
+ {
+ Objects.requireNonNull(fingerprint, "fingerprint");
+ Objects.requireNonNull(selectedBundleNames, "selectedBundleNames");
+ final Set<String> newSelection = Set.copyOf(selectedBundleNames);
+
+ // Phase 1: snapshot under entryLock and compute the diff. Do NOT call
StorageLocation methods here — that would
+ // acquire the location's readLock while holding entryLock and could
deadlock with a concurrent writeLock holder
+ // that needs entryLock
+ final StorageLocation loc;
+ final boolean needsSelfHold;
+ final List<String> namesToRelease;
+ final List<String> namesToAcquire;
+ entryLock.lock();
+ try {
+ if (location == null) {
+ throw DruidException.defensive(
+ "applyRule on partial metadata entry[%s] requires the entry to be
mounted",
+ id
+ );
+ }
+ if (fingerprint.equals(ruleFingerprint) &&
newSelection.equals(ruleSelectedBundleNames)) {
+ return;
+ }
+ loc = location;
+ needsSelfHold = (metadataSelfHold == null);
+ namesToRelease = new ArrayList<>();
+ for (String name : ruleBundleHolds.keySet()) {
+ if (!newSelection.contains(name)) {
+ namesToRelease.add(name);
+ }
+ }
+ namesToAcquire = new ArrayList<>();
+ for (String name : newSelection) {
+ if (!ruleBundleHolds.containsKey(name) && findLinkedBundleByName(name)
!= null) {
+ namesToAcquire.add(name);
+ }
+ }
+ }
+ finally {
+ entryLock.unlock();
+ }
+
+ // Phase 2 + 3: acquire outside entryLock, then commit under entryLock.
Track every hold this call acquires in
+ // `uncommittedHolds`; when a hold's ownership transfers into
`metadataSelfHold` or `ruleBundleHolds`, it is
+ // removed from the list. Any throw between acquire and commit (or a
race-lose at commit) leaves the hold in
+ // `uncommittedHolds`, and the finally at Phase 4 releases it. This is
what makes applyRule all-or-nothing at
+ // the ReservationHold level: no leaks on partial failure, no stranded
self-hold under a stale fingerprint.
+ final List<StorageLocation.ReservationHold<?>> uncommittedHolds = new
ArrayList<>();
+ final List<StorageLocation.ReservationHold<?>> displacedHolds = new
ArrayList<>();
+ try {
+ final StorageLocation.ReservationHold<PartialSegmentMetadataCacheEntry>
newSelfHold;
+ if (needsSelfHold) {
+ newSelfHold = loc.addWeakReservationHoldIfExists(id);
+ if (newSelfHold == null) {
+ throw DruidException.defensive(
+ "Failed to acquire self-referential rule-hold on partial
metadata entry[%s]; entry is not weak-reserved",
+ id
+ );
+ }
+ uncommittedHolds.add(newSelfHold);
+ } else {
+ newSelfHold = null;
+ }
+ final Map<String,
StorageLocation.ReservationHold<PartialSegmentBundleCacheEntry>> acquired = new
HashMap<>();
+ for (String name : namesToAcquire) {
+ final StorageLocation.ReservationHold<PartialSegmentBundleCacheEntry>
h =
+ loc.addWeakReservationHoldIfExists(new
PartialSegmentBundleCacheEntryIdentifier(segmentId, name));
+ if (h != null) {
+ acquired.put(name, h);
+ uncommittedHolds.add(h);
+ }
+ }
+
+ // Phase 3: commit under entryLock. Ownership transfers are done by
removing from `uncommittedHolds` after
+ // installing into the field. Anything left in `uncommittedHolds` at the
end lost a race and gets released
+ // in Phase 4 alongside `displacedHolds` (the pre-existing holds diffed
out of ruleBundleHolds).
+ entryLock.lock();
+ try {
+ if (newSelfHold != null) {
+ if (metadataSelfHold == null) {
+ metadataSelfHold = newSelfHold;
+ uncommittedHolds.remove(newSelfHold);
+ }
+ // else: a concurrent applyRule installed a self-hold (shouldn't
happen under segmentLock, but defensive).
+ // newSelfHold stays in uncommittedHolds, gets released in Phase 4.
Review Comment:
so this "else" is just describing why the above if exists versus blindly
re-assigning `metadataSelfHold`? the comment wording here and on L408 kinda
give off vibes of a non-existent else block that it is documenting
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java:
##########
@@ -1066,15 +1435,71 @@ public File getSegmentFiles(final DataSegment segment)
@Override
public void drop(final DataSegment segment)
{
- final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(segment.getId());
- for (StorageLocation location : locations) {
- final CacheEntry entry = location.getCacheEntry(id);
- if (entry != null) {
- location.release(entry);
+ final ReferenceCountingLock lock = lock(segment);
+ synchronized (lock) {
+ try {
+ // partial-load-rule cleanup: if the segment's cache entry is a
partial metadata entry with an applied rule,
+ // clear it
+ final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(segment.getId());
+ for (StorageLocation location : locations) {
+ final CacheEntry entry = location.getCacheEntry(id);
+ if (entry instanceof PartialSegmentMetadataCacheEntry partial) {
+ partial.clearRule();
+ // Force synchronous cleanup: clearRule releases the self-hold,
but the info-file-deletion hook only
+ // fires on doActualUnmount which is triggered by the phaser
hitting zero. removeUnheldWeakEntry
+ // unlinks the weak entry from cache and terminates the phaser
now, firing the hook (and the mapper
+ // teardown) on the caller's thread. No-op if a concurrent query
still holds the entry.
+ location.removeUnheldWeakEntry(id);
+ }
+ }
+ // full-segment drop path: release any {@link
CompleteSegmentCacheEntry} still reserved.
+ for (StorageLocation location : locations) {
+ final CacheEntry entry = location.getCacheEntry(id);
+ if (entry != null) {
+ location.release(entry);
+ }
+ }
Review Comment:
can we fold this in to the first for loop over locations??
##########
server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java:
##########
@@ -0,0 +1,666 @@
+/*
+ * 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.druid.segment.loading;
+
+import com.fasterxml.jackson.databind.InjectableValues;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.ListBasedInputRow;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.guice.LocalDataStorageDruidModule;
+import org.apache.druid.jackson.SegmentizerModule;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.query.expression.TestExprMacroTable;
+import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.IndexIO;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.SegmentLazyLoadFailCallback;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.column.ColumnConfig;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
+import org.apache.druid.segment.incremental.IncrementalIndexSchema;
+import org.apache.druid.segment.projections.Projections;
+import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.partition.NoneShardSpec;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Manager-level tests for the partial-load-rule {@link
SegmentLocalCacheManager#load} path: a segment whose
+ * {@code loadSpec} is a {@link PartialLoadSpec} wrapper drives
+ * {@link PartialSegmentMetadataCacheEntry#applyRule} on the metadata entry,
which installs a self-referential
+ * {@link StorageLocation.ReservationHold} to keep the metadata resident and
holds on each selected bundle already
+ * registered. Selected bundles not yet registered get eager-downloads
submitted to the loading pool; {@code load}
+ * blocks until every eager download completes so the announced fingerprint
reflects "rule fully realized". On any
+ * eager-download failure the rule state is cleared and a {@link
SegmentLoadingException} propagates so the
+ * coordinator's load queue retries. The cache entries themselves live in
{@code weakCacheEntries} exactly as they
+ * would from an on-demand acquire; the rule-holds keep them from being
SIEVE-evicted for as long as the coordinator
+ * considers the segment loaded.
+ * <p>
+ * {@code drop} calls {@link PartialSegmentMetadataCacheEntry#clearRule} —
entries become unheld weak (subject to any
+ * remaining query holds) and SIEVE reclaims them naturally.
+ */
+class SegmentLocalCacheManagerPartialRuleLoadTest
+{
+ private static final SegmentId SEGMENT_ID = SegmentId.of("test",
Intervals.of("2025/2026"), "v1", 0);
+ private static final DateTime TIME = DateTimes.of("2025-01-01");
+ private static final String AGG_BUNDLE = "dim1_metric1_sum";
+ private static final String OTHER_AGG_BUNDLE = "dim1_count";
+ private static final String FINGERPRINT = "v1:rule-bundle-test";
+
+ private static final RowSignature ROW_SIGNATURE = RowSignature.builder()
+ .add("dim1",
ColumnType.STRING)
+
.add("metric1", ColumnType.LONG)
+ .build();
+
+ private static final List<AggregateProjectionSpec> PROJECTIONS =
Arrays.asList(
+ AggregateProjectionSpec.builder(AGG_BUNDLE)
+ .groupingColumns(new
StringDimensionSchema("dim1"))
+ .aggregators(
+ new LongSumAggregatorFactory("_metric1_sum",
"metric1"),
+ new CountAggregatorFactory("_count")
+ )
+ .build(),
+ // A second aggregate projection so tests can verify that non-selected
bundles are NOT pre-mounted while
+ // selected + parent bundles ARE.
+ AggregateProjectionSpec.builder(OTHER_AGG_BUNDLE)
+ .groupingColumns(new
StringDimensionSchema("dim1"))
+ .aggregators(new CountAggregatorFactory("_count"))
+ .build()
+ );
+
+ private static final List<InputRow> ROWS = Arrays.asList(
+ new ListBasedInputRow(ROW_SIGNATURE, TIME,
ROW_SIGNATURE.getColumnNames(), Arrays.asList("a", 1L)),
+ new ListBasedInputRow(ROW_SIGNATURE, TIME.plusMinutes(1),
ROW_SIGNATURE.getColumnNames(), Arrays.asList("a", 2L)),
+ new ListBasedInputRow(ROW_SIGNATURE, TIME.plusMinutes(2),
ROW_SIGNATURE.getColumnNames(), Arrays.asList("b", 3L)),
+ new ListBasedInputRow(ROW_SIGNATURE, TIME.plusMinutes(3),
ROW_SIGNATURE.getColumnNames(), Arrays.asList("b", 4L))
+ );
+
+ @TempDir
+ static File SHARED_TEMP_DIR;
+
+ private static File DEEP_STORAGE_DIR;
+
+ @TempDir
+ File perTestTempDir;
+
+ private ObjectMapper jsonMapper;
+ private File cacheRoot;
+ private SegmentLocalCacheManager manager;
+
+ @BeforeAll
+ static void buildSegment()
+ {
+ final File tmp = new File(SHARED_TEMP_DIR, "build_" +
ThreadLocalRandom.current().nextInt());
+ DEEP_STORAGE_DIR = IndexBuilder.create()
+ .useV10()
+ .tmpDir(tmp)
+
.segmentWriteOutMediumFactory(OffHeapMemorySegmentWriteOutMediumFactory.instance())
+ .schema(
+ IncrementalIndexSchema.builder()
+
.withDimensionsSpec(
+
DimensionsSpec.builder()
+
.setDimensions(
+
List.of(
+
new StringDimensionSchema("dim1"),
+
new LongDimensionSchema("metric1")
+
)
+
)
+
.build()
+ )
+ .withRollup(false)
+
.withMinTimestamp(TIME.getMillis())
+
.withProjections(PROJECTIONS)
+ .build()
+ )
+ .indexSpec(IndexSpec.builder()
+
.withMetadataCompression(CompressionStrategy.NONE)
+ .build())
+ .rows(ROWS)
+ .buildMMappedIndexFile();
+ EmittingLogger.registerEmitter(new NoopServiceEmitter());
+ }
+
+ @BeforeEach
+ void setup() throws IOException
+ {
+ jsonMapper = TestHelper.makeJsonMapper();
+ jsonMapper.registerSubtypes(new NamedType(LocalLoadSpec.class, "local"));
+ jsonMapper.registerSubtypes(new NamedType(PartialProjectionLoadSpec.class,
PartialProjectionLoadSpec.TYPE));
+ jsonMapper.registerModule(new SegmentizerModule());
+ jsonMapper.registerModules(new
LocalDataStorageDruidModule().getJacksonModules());
+ jsonMapper.setInjectableValues(
+ new InjectableValues.Std()
+ .addValue(LocalDataSegmentPuller.class, new
LocalDataSegmentPuller())
+ .addValue(IndexIO.class, TestHelper.getTestIndexIO(jsonMapper,
ColumnConfig.DEFAULT))
+ .addValue(ObjectMapper.class, jsonMapper)
+ .addValue(DataSegment.PruneSpecsHolder.class,
DataSegment.PruneSpecsHolder.DEFAULT)
+ .addValue(ExprMacroTable.class, TestExprMacroTable.INSTANCE)
+ );
+
+ cacheRoot = new File(perTestTempDir, "cache_" +
ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE));
+ FileUtils.mkdirp(cacheRoot);
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ if (manager != null) {
+ // Drop the segment to release rule-holds and unmount mmap'd bundle
files before @TempDir tries to clean up
+ // the cache directory. Without this, on some platforms the temp-dir
cleanup fails on still-mapped files.
+ try {
+ manager.drop(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ }
+ catch (Throwable ignored) {
+ // best-effort — some tests never installed a rule on this segment
+ }
+ manager.shutdown();
+ }
+ }
+
+ @Test
+ void testLoadInstallsRuleHoldsOnMetadataAndSelectedBundle() throws Exception
+ {
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ // The metadata entry now carries the rule state.
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertTrue(metadata.isRuleHeld(), "rule must be applied to the
metadata entry");
+ Assertions.assertEquals(FINGERPRINT, metadata.getRuleFingerprint(),
"fingerprint must be stored on the entry");
+ // Metadata + selected bundle + its base parent are weak-reserved on the
location; NOT static.
+ final SegmentCacheEntryIdentifier metaId = new
SegmentCacheEntryIdentifier(SEGMENT_ID);
+ final PartialSegmentBundleCacheEntryIdentifier aggId =
+ new PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID, AGG_BUNDLE);
+ final PartialSegmentBundleCacheEntryIdentifier baseId =
+ new PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID,
Projections.BASE_TABLE_PROJECTION_NAME);
+ Assertions.assertTrue(location.isWeakReserved(metaId), "metadata entry
should be weak-reserved");
+ Assertions.assertTrue(location.isWeakReserved(aggId), "selected bundle
should be weak-reserved");
+ Assertions.assertTrue(location.isWeakReserved(baseId), "base dependency
should be weak-reserved");
+ Assertions.assertFalse(location.isReserved(metaId), "metadata entry should
NOT be in staticCacheEntries");
+ Assertions.assertFalse(location.isReserved(aggId), "selected bundle should
NOT be in staticCacheEntries");
+ }
+
+ @Test
+ void testLoadDoesNotReserveNonSelectedBundles() throws Exception
+ {
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ final PartialSegmentBundleCacheEntryIdentifier otherAggId =
+ new PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID,
OTHER_AGG_BUNDLE);
+ Assertions.assertFalse(
+ location.isWeakReserved(otherAggId),
+ "non-selected bundle must not be reserved by loadPartial"
+ );
+ Assertions.assertNull(
+ location.getCacheEntry(otherAggId),
+ "non-selected bundle must not be in the cache at all after load"
+ );
+ }
+
+ @Test
+ void testLoadEagerlyDownloadsSelectedBundleContainers() throws Exception
+ {
+ // ensureBundleDownloaded should be a no-op after load — the
eager-download pool task already downloaded the
+ // bundle.
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ final PartialSegmentFileMapperV10 mapper = metadata.getFileMapper();
+ Assertions.assertNotNull(mapper, "metadata mount should produce a file
mapper");
+ final long downloadedBefore = mapper.getDownloadedBytes();
+ mapper.ensureBundleDownloaded(AGG_BUNDLE);
+ Assertions.assertEquals(
+ downloadedBefore,
+ mapper.getDownloadedBytes(),
+ "ensureBundleDownloaded must be a no-op — the eager task already
downloaded the bundle"
+ );
+ }
+
+ @Test
+ void testLoadWithoutVirtualStorageDoesNotInstallRuleHolds() throws Exception
+ {
+ // virtualStorage=false: partial-load rules aren't a concept here. load()
takes the eager full-load path (via
+ // PartialLoadSpec.loadSegment → the materialized inner LocalLoadSpec),
and no rule state is created.
+ manager = makeManager(false, false);
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ Assertions.assertNull(
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "no rule state on the eager path"
+ );
+ }
+
+ @Test
+ void testLoadWithPartialsDisabledIsNoop() throws Exception
+ {
+ manager = makeManager(true, false);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ Assertions.assertNull(
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "with partial downloads disabled, loadPartial must not run"
+ );
+ Assertions.assertFalse(
+ location.isWeakReserved(new SegmentCacheEntryIdentifier(SEGMENT_ID)),
+ "with partial downloads disabled, no reservation should be installed
at load time"
+ );
+ }
+
+ @Test
+ void testSameFingerprintReIssueIsIdempotent() throws Exception
+ {
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ final PartialSegmentMetadataCacheEntry firstMeta =
weakReservedMetadata(location, SEGMENT_ID);
+
+ // Second load with the same fingerprint must be idempotent — same
metadata entry, unchanged rule state.
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ final PartialSegmentMetadataCacheEntry secondMeta =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertSame(firstMeta, secondMeta, "same-fingerprint re-issue
must not create a new metadata entry");
+ Assertions.assertEquals(FINGERPRINT, secondMeta.getRuleFingerprint(),
"fingerprint unchanged after re-issue");
+ }
+
+ @Test
+ void testRuleChangeSwapsHoldsWithoutDroppingSharedMetadata() throws Exception
+ {
+ // Same segment, different fingerprint: the rule's bundle selection
changes but the underlying metadata entry is
+ // the same weak entry (its self-hold is never released during the swap;
only bundle rule-holds diff).
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE),
"v1:rule-original"));
+ final PartialSegmentMetadataCacheEntry meta =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertEquals("v1:rule-original", meta.getRuleFingerprint());
+
+ manager.load(partialWrapperSegment(List.of(OTHER_AGG_BUNDLE),
"v2:rule-updated"));
+
+ // Metadata entry is the SAME instance across the swap.
+ final PartialSegmentMetadataCacheEntry metaAfter =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertSame(meta, metaAfter, "rule change must reuse the
underlying metadata entry");
+ Assertions.assertEquals("v2:rule-updated", metaAfter.getRuleFingerprint(),
"fingerprint must swap");
+ // New rule's bundle is rule-held.
+ Assertions.assertTrue(
+ location.isWeakReserved(new
PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID, OTHER_AGG_BUNDLE)),
+ "new rule's bundle should be reserved"
+ );
+ }
+
+ @Test
+ void testDropClearsRule() throws Exception
+ {
+ manager = makeManager(true, true);
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ Assertions.assertNotNull(manager.getRuleFingerprintForSegment(SEGMENT_ID));
+
+ manager.drop(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ Assertions.assertNull(
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "drop must clear the rule state on the metadata entry"
+ );
+ }
+
+ @Test
+ void testRangeReaderNullDropsPriorRule() throws Exception
+ {
+ // Prior rule installed. Second load with a wrapper whose delegate can't
produce a range reader releases the rule
+ // via clearRule so subsequent queries fall through to weak-full-load
semantics.
+ manager = makeManager(true, true);
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE),
"v1:rule-original"));
+ Assertions.assertNotNull(manager.getRuleFingerprintForSegment(SEGMENT_ID));
+
+ manager.load(partialWrapperSegmentWithNullRangeReader(List.of(AGG_BUNDLE),
"v2:rule-updated"));
+
+ Assertions.assertNull(
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "null range reader must clear the prior rule"
+ );
+ }
+
+ @Test
+ void testRangeReaderNullNoopWhenNoPriorRule() throws Exception
+ {
+ // No prior rule, no range reader → no rule installed, but also no
failure. Segment simply doesn't get a rule;
+ // a later query would drive weak-full-load via the on-demand path.
+ manager = makeManager(true, true);
+ manager.load(partialWrapperSegmentWithNullRangeReader(List.of(AGG_BUNDLE),
"v1:rule-noop"));
+ Assertions.assertNull(manager.getRuleFingerprintForSegment(SEGMENT_ID));
+ }
+
+ @Test
+ void testLoadFailureLeavesNoRuleApplied() throws Exception
+ {
+ // Wrapper referring to a non-existent projection —
wrapper.getSelectedBundleNames throws before applyRule fires.
+ // In the v2 design there is no rollback-nuke of the partial dir on
failure; the transient hold releases and the
+ // now-unheld weak metadata entry becomes SIEVE-eligible naturally
(info-file cleanup fires via the entry's
+ // onUnmount hook when SIEVE eventually reclaims it). The important
correctness invariant is: no stranded rule
+ // state, no partial rule fingerprint applied.
+ manager = makeManager(true, true);
+ final DataSegment segment =
partialWrapperSegment(List.of("does_not_exist"));
+
+ final Throwable thrown = Assertions.assertThrows(
+ Throwable.class,
+ () -> manager.load(segment)
+ );
+ Assertions.assertTrue(
+ thrown.getMessage() != null && thrown.getMessage().contains("does not
contain projection[does_not_exist]")
+ || thrown.getCause() != null
+ && thrown.getCause().getMessage().contains("does not contain
projection[does_not_exist]"),
+ "unexpected throwable: " + thrown
+ );
+ Assertions.assertNull(
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "failed load must not leave a rule applied on the metadata entry"
+ );
+ }
+
+ @Test
+ void testBootstrapReinstallsRuleHoldsFromPersistedInfoFile() throws Exception
+ {
+ // Regression guard for the "brief post-bootstrap SIEVE-eligible window"
concern: on historical restart,
Review Comment:
```suggestion
// on historical restart,
```
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java:
##########
@@ -1066,15 +1435,71 @@ public File getSegmentFiles(final DataSegment segment)
@Override
public void drop(final DataSegment segment)
{
- final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(segment.getId());
- for (StorageLocation location : locations) {
- final CacheEntry entry = location.getCacheEntry(id);
- if (entry != null) {
- location.release(entry);
+ final ReferenceCountingLock lock = lock(segment);
+ synchronized (lock) {
+ try {
+ // partial-load-rule cleanup: if the segment's cache entry is a
partial metadata entry with an applied rule,
+ // clear it
+ final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(segment.getId());
+ for (StorageLocation location : locations) {
+ final CacheEntry entry = location.getCacheEntry(id);
+ if (entry instanceof PartialSegmentMetadataCacheEntry partial) {
+ partial.clearRule();
+ // Force synchronous cleanup: clearRule releases the self-hold,
but the info-file-deletion hook only
+ // fires on doActualUnmount which is triggered by the phaser
hitting zero. removeUnheldWeakEntry
+ // unlinks the weak entry from cache and terminates the phaser
now, firing the hook (and the mapper
+ // teardown) on the caller's thread. No-op if a concurrent query
still holds the entry.
+ location.removeUnheldWeakEntry(id);
Review Comment:
I'm confused how this work and actually force full cleanup if we don't drive
bundle unmounts before we call in to this?
##########
server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java:
##########
@@ -0,0 +1,666 @@
+/*
+ * 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.druid.segment.loading;
+
+import com.fasterxml.jackson.databind.InjectableValues;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.ListBasedInputRow;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.guice.LocalDataStorageDruidModule;
+import org.apache.druid.jackson.SegmentizerModule;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.query.expression.TestExprMacroTable;
+import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.IndexIO;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.SegmentLazyLoadFailCallback;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.column.ColumnConfig;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
+import org.apache.druid.segment.incremental.IncrementalIndexSchema;
+import org.apache.druid.segment.projections.Projections;
+import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.partition.NoneShardSpec;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Manager-level tests for the partial-load-rule {@link
SegmentLocalCacheManager#load} path: a segment whose
+ * {@code loadSpec} is a {@link PartialLoadSpec} wrapper drives
+ * {@link PartialSegmentMetadataCacheEntry#applyRule} on the metadata entry,
which installs a self-referential
+ * {@link StorageLocation.ReservationHold} to keep the metadata resident and
holds on each selected bundle already
+ * registered. Selected bundles not yet registered get eager-downloads
submitted to the loading pool; {@code load}
+ * blocks until every eager download completes so the announced fingerprint
reflects "rule fully realized". On any
+ * eager-download failure the rule state is cleared and a {@link
SegmentLoadingException} propagates so the
+ * coordinator's load queue retries. The cache entries themselves live in
{@code weakCacheEntries} exactly as they
+ * would from an on-demand acquire; the rule-holds keep them from being
SIEVE-evicted for as long as the coordinator
+ * considers the segment loaded.
+ * <p>
+ * {@code drop} calls {@link PartialSegmentMetadataCacheEntry#clearRule} —
entries become unheld weak (subject to any
+ * remaining query holds) and SIEVE reclaims them naturally.
+ */
+class SegmentLocalCacheManagerPartialRuleLoadTest
+{
+ private static final SegmentId SEGMENT_ID = SegmentId.of("test",
Intervals.of("2025/2026"), "v1", 0);
+ private static final DateTime TIME = DateTimes.of("2025-01-01");
+ private static final String AGG_BUNDLE = "dim1_metric1_sum";
+ private static final String OTHER_AGG_BUNDLE = "dim1_count";
+ private static final String FINGERPRINT = "v1:rule-bundle-test";
+
+ private static final RowSignature ROW_SIGNATURE = RowSignature.builder()
+ .add("dim1",
ColumnType.STRING)
+
.add("metric1", ColumnType.LONG)
+ .build();
+
+ private static final List<AggregateProjectionSpec> PROJECTIONS =
Arrays.asList(
+ AggregateProjectionSpec.builder(AGG_BUNDLE)
+ .groupingColumns(new
StringDimensionSchema("dim1"))
+ .aggregators(
+ new LongSumAggregatorFactory("_metric1_sum",
"metric1"),
+ new CountAggregatorFactory("_count")
+ )
+ .build(),
+ // A second aggregate projection so tests can verify that non-selected
bundles are NOT pre-mounted while
+ // selected + parent bundles ARE.
+ AggregateProjectionSpec.builder(OTHER_AGG_BUNDLE)
+ .groupingColumns(new
StringDimensionSchema("dim1"))
+ .aggregators(new CountAggregatorFactory("_count"))
+ .build()
+ );
+
+ private static final List<InputRow> ROWS = Arrays.asList(
+ new ListBasedInputRow(ROW_SIGNATURE, TIME,
ROW_SIGNATURE.getColumnNames(), Arrays.asList("a", 1L)),
+ new ListBasedInputRow(ROW_SIGNATURE, TIME.plusMinutes(1),
ROW_SIGNATURE.getColumnNames(), Arrays.asList("a", 2L)),
+ new ListBasedInputRow(ROW_SIGNATURE, TIME.plusMinutes(2),
ROW_SIGNATURE.getColumnNames(), Arrays.asList("b", 3L)),
+ new ListBasedInputRow(ROW_SIGNATURE, TIME.plusMinutes(3),
ROW_SIGNATURE.getColumnNames(), Arrays.asList("b", 4L))
+ );
+
+ @TempDir
+ static File SHARED_TEMP_DIR;
+
+ private static File DEEP_STORAGE_DIR;
+
+ @TempDir
+ File perTestTempDir;
+
+ private ObjectMapper jsonMapper;
+ private File cacheRoot;
+ private SegmentLocalCacheManager manager;
+
+ @BeforeAll
+ static void buildSegment()
+ {
+ final File tmp = new File(SHARED_TEMP_DIR, "build_" +
ThreadLocalRandom.current().nextInt());
+ DEEP_STORAGE_DIR = IndexBuilder.create()
+ .useV10()
+ .tmpDir(tmp)
+
.segmentWriteOutMediumFactory(OffHeapMemorySegmentWriteOutMediumFactory.instance())
+ .schema(
+ IncrementalIndexSchema.builder()
+
.withDimensionsSpec(
+
DimensionsSpec.builder()
+
.setDimensions(
+
List.of(
+
new StringDimensionSchema("dim1"),
+
new LongDimensionSchema("metric1")
+
)
+
)
+
.build()
+ )
+ .withRollup(false)
+
.withMinTimestamp(TIME.getMillis())
+
.withProjections(PROJECTIONS)
+ .build()
+ )
+ .indexSpec(IndexSpec.builder()
+
.withMetadataCompression(CompressionStrategy.NONE)
+ .build())
+ .rows(ROWS)
+ .buildMMappedIndexFile();
+ EmittingLogger.registerEmitter(new NoopServiceEmitter());
+ }
+
+ @BeforeEach
+ void setup() throws IOException
+ {
+ jsonMapper = TestHelper.makeJsonMapper();
+ jsonMapper.registerSubtypes(new NamedType(LocalLoadSpec.class, "local"));
+ jsonMapper.registerSubtypes(new NamedType(PartialProjectionLoadSpec.class,
PartialProjectionLoadSpec.TYPE));
+ jsonMapper.registerModule(new SegmentizerModule());
+ jsonMapper.registerModules(new
LocalDataStorageDruidModule().getJacksonModules());
+ jsonMapper.setInjectableValues(
+ new InjectableValues.Std()
+ .addValue(LocalDataSegmentPuller.class, new
LocalDataSegmentPuller())
+ .addValue(IndexIO.class, TestHelper.getTestIndexIO(jsonMapper,
ColumnConfig.DEFAULT))
+ .addValue(ObjectMapper.class, jsonMapper)
+ .addValue(DataSegment.PruneSpecsHolder.class,
DataSegment.PruneSpecsHolder.DEFAULT)
+ .addValue(ExprMacroTable.class, TestExprMacroTable.INSTANCE)
+ );
+
+ cacheRoot = new File(perTestTempDir, "cache_" +
ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE));
+ FileUtils.mkdirp(cacheRoot);
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ if (manager != null) {
+ // Drop the segment to release rule-holds and unmount mmap'd bundle
files before @TempDir tries to clean up
+ // the cache directory. Without this, on some platforms the temp-dir
cleanup fails on still-mapped files.
+ try {
+ manager.drop(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ }
+ catch (Throwable ignored) {
+ // best-effort — some tests never installed a rule on this segment
+ }
+ manager.shutdown();
+ }
+ }
+
+ @Test
+ void testLoadInstallsRuleHoldsOnMetadataAndSelectedBundle() throws Exception
+ {
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ // The metadata entry now carries the rule state.
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertTrue(metadata.isRuleHeld(), "rule must be applied to the
metadata entry");
+ Assertions.assertEquals(FINGERPRINT, metadata.getRuleFingerprint(),
"fingerprint must be stored on the entry");
+ // Metadata + selected bundle + its base parent are weak-reserved on the
location; NOT static.
+ final SegmentCacheEntryIdentifier metaId = new
SegmentCacheEntryIdentifier(SEGMENT_ID);
+ final PartialSegmentBundleCacheEntryIdentifier aggId =
+ new PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID, AGG_BUNDLE);
+ final PartialSegmentBundleCacheEntryIdentifier baseId =
+ new PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID,
Projections.BASE_TABLE_PROJECTION_NAME);
+ Assertions.assertTrue(location.isWeakReserved(metaId), "metadata entry
should be weak-reserved");
+ Assertions.assertTrue(location.isWeakReserved(aggId), "selected bundle
should be weak-reserved");
+ Assertions.assertTrue(location.isWeakReserved(baseId), "base dependency
should be weak-reserved");
+ Assertions.assertFalse(location.isReserved(metaId), "metadata entry should
NOT be in staticCacheEntries");
+ Assertions.assertFalse(location.isReserved(aggId), "selected bundle should
NOT be in staticCacheEntries");
+ }
+
+ @Test
+ void testLoadDoesNotReserveNonSelectedBundles() throws Exception
+ {
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ final PartialSegmentBundleCacheEntryIdentifier otherAggId =
+ new PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID,
OTHER_AGG_BUNDLE);
+ Assertions.assertFalse(
+ location.isWeakReserved(otherAggId),
+ "non-selected bundle must not be reserved by loadPartial"
+ );
+ Assertions.assertNull(
+ location.getCacheEntry(otherAggId),
+ "non-selected bundle must not be in the cache at all after load"
+ );
+ }
+
+ @Test
+ void testLoadEagerlyDownloadsSelectedBundleContainers() throws Exception
+ {
+ // ensureBundleDownloaded should be a no-op after load — the
eager-download pool task already downloaded the
+ // bundle.
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ final PartialSegmentFileMapperV10 mapper = metadata.getFileMapper();
+ Assertions.assertNotNull(mapper, "metadata mount should produce a file
mapper");
+ final long downloadedBefore = mapper.getDownloadedBytes();
+ mapper.ensureBundleDownloaded(AGG_BUNDLE);
+ Assertions.assertEquals(
+ downloadedBefore,
+ mapper.getDownloadedBytes(),
+ "ensureBundleDownloaded must be a no-op — the eager task already
downloaded the bundle"
+ );
+ }
+
+ @Test
+ void testLoadWithoutVirtualStorageDoesNotInstallRuleHolds() throws Exception
+ {
+ // virtualStorage=false: partial-load rules aren't a concept here. load()
takes the eager full-load path (via
+ // PartialLoadSpec.loadSegment → the materialized inner LocalLoadSpec),
and no rule state is created.
+ manager = makeManager(false, false);
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ Assertions.assertNull(
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "no rule state on the eager path"
+ );
+ }
+
+ @Test
+ void testLoadWithPartialsDisabledIsNoop() throws Exception
+ {
+ manager = makeManager(true, false);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+
+ Assertions.assertNull(
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "with partial downloads disabled, loadPartial must not run"
+ );
+ Assertions.assertFalse(
+ location.isWeakReserved(new SegmentCacheEntryIdentifier(SEGMENT_ID)),
+ "with partial downloads disabled, no reservation should be installed
at load time"
+ );
+ }
+
+ @Test
+ void testSameFingerprintReIssueIsIdempotent() throws Exception
+ {
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ final PartialSegmentMetadataCacheEntry firstMeta =
weakReservedMetadata(location, SEGMENT_ID);
+
+ // Second load with the same fingerprint must be idempotent — same
metadata entry, unchanged rule state.
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE)));
+ final PartialSegmentMetadataCacheEntry secondMeta =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertSame(firstMeta, secondMeta, "same-fingerprint re-issue
must not create a new metadata entry");
+ Assertions.assertEquals(FINGERPRINT, secondMeta.getRuleFingerprint(),
"fingerprint unchanged after re-issue");
+ }
+
+ @Test
+ void testRuleChangeSwapsHoldsWithoutDroppingSharedMetadata() throws Exception
+ {
+ // Same segment, different fingerprint: the rule's bundle selection
changes but the underlying metadata entry is
+ // the same weak entry (its self-hold is never released during the swap;
only bundle rule-holds diff).
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE),
"v1:rule-original"));
+ final PartialSegmentMetadataCacheEntry meta =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertEquals("v1:rule-original", meta.getRuleFingerprint());
+
+ manager.load(partialWrapperSegment(List.of(OTHER_AGG_BUNDLE),
"v2:rule-updated"));
+
+ // Metadata entry is the SAME instance across the swap.
+ final PartialSegmentMetadataCacheEntry metaAfter =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertSame(meta, metaAfter, "rule change must reuse the
underlying metadata entry");
+ Assertions.assertEquals("v2:rule-updated", metaAfter.getRuleFingerprint(),
"fingerprint must swap");
+ // New rule's bundle is rule-held.
Review Comment:
should AGG_BUNDLE be asserted as not rule held?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
