FrankChen021 commented on code in PR #19657:
URL: https://github.com/apache/druid/pull/19657#discussion_r3536420944
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java:
##########
@@ -1069,9 +1659,18 @@ 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);
+ if (entry == null) {
+ continue;
}
+ // Cascade-release linked bundle entries before the metadata. {@link
StorageLocation#release} only operates on
+ // entries in {@code staticCacheEntries}, so this is a no-op for
weak/on-demand bundles; for static rule-pinned
+ // bundles installed by {@link #loadPartial}, this is the only path that
removes them from the static map.
+ if (entry instanceof PartialSegmentMetadataCacheEntry partial) {
+ for (PartialSegmentBundleCacheEntry bundle :
partial.snapshotLinkedBundles()) {
+ location.release(bundle);
Review Comment:
[P2] Weak linked bundles survive partial drop
`location.release(bundle)` only removes entries from `staticCacheEntries`,
so it is a no-op for weak/on-demand bundles linked to a static partial metadata
entry. Those weak bundles still hold metadata references, but the
reconciliation precheck subtracts all linked bundles as if drop will release
them, after which the new code can nuke or reuse the segment directory while
the weak bundles remain mounted. Drop/reconcile should either remove unheld
weak linked bundles or treat their presence as a deferred-cleanup blocker.
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java:
##########
@@ -872,6 +925,493 @@ private record ReservedPartial(
{
}
+ /**
+ * Handle the partial-load-rule case at {@link #load} time: reserve a {@link
PartialSegmentMetadataCacheEntry} as
+ * static, persist the info file, mount the metadata header, and
reserve+mount each rule-selected bundle as static
+ * with its data eagerly downloaded (non-selected bundles are downloaded
on-demand at query time via the existing
+ * weak path).
+ * <p>
+ * The entire flow runs synchronously on {@link
#virtualStorageLoadingThreadPool} (via submit + get) so it respects
+ * the pool's concurrency cap and virtual-thread runtime while giving
+ * {@link org.apache.druid.server.coordination.SegmentLoadDropHandler} a
truthful success/failure signal. On any
+ * failure (mount, bundle reservation, download) the metadata + any
partially-reserved bundles are cascade-released
+ * before the {@link SegmentLoadingException} propagates, so no orphan
static reservation survives the load call.
+ */
+ private void loadPartial(DataSegment dataSegment) throws
SegmentLoadingException
+ {
+ final PartialLoadSpec wrapper = materializePartialLoadSpec(dataSegment);
+ final SegmentRangeReader rangeReader = openPartialRangeReader(dataSegment,
wrapper);
+ if (rangeReader == null) {
+ handleUnsupportedRangeReaderFallback(dataSegment, wrapper);
+ return;
+ }
+ final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(dataSegment.getId());
+ final ReferenceCountingLock lock = lock(dataSegment);
+ synchronized (lock) {
+ try {
+ if (reconcileExistingEntries(dataSegment, wrapper, id)) {
+ return;
+ }
+ final PartialReservation reservation =
+ reserveMetadataOnFirstAvailableLocation(dataSegment, wrapper,
rangeReader);
+ persistInfoFileAndAttachCleanupHook(dataSegment, reservation);
+ mountAndAwaitPartial(dataSegment, wrapper, reservation);
+ }
+ finally {
+ unlock(dataSegment, lock);
+ }
+ }
+ }
+
+ /**
+ * Materialize the segment's wrapped load spec to a {@link PartialLoadSpec}.
Fails with a diagnostic when Jackson
+ * can't convert (missing subtype registration, malformed wire form) or when
the materialized type isn't a partial
+ * wrapper despite the dispatch check upstream.
+ */
+ private PartialLoadSpec materializePartialLoadSpec(DataSegment dataSegment)
throws SegmentLoadingException
+ {
+ final LoadSpec materializedLoadSpec;
+ try {
+ materializedLoadSpec =
jsonMapper.convertValue(dataSegment.getLoadSpec(), LoadSpec.class);
+ }
+ catch (Exception e) {
+ throw new SegmentLoadingException(
+ e,
+ "Failed to materialize partial load spec for segment[%s]",
+ dataSegment.getId()
+ );
+ }
+ if (!(materializedLoadSpec instanceof PartialLoadSpec wrapper)) {
+ throw DruidException.defensive(
+ "Segment[%s] load spec was detected as partial but materialized to
non-partial type[%s]",
+ dataSegment.getId(),
+ materializedLoadSpec.getClass().getSimpleName()
+ );
+ }
+ return wrapper;
+ }
+
+ /**
+ * Open a range reader for the segment's deep storage, wrapping any {@link
IOException} as a
+ * {@link SegmentLoadingException}. Returns {@code null} if the backend
doesn't support range reads (e.g. zipped
+ * deep storage) so caller falls back to weak-full-load semantics.
+ */
+ @Nullable
+ private SegmentRangeReader openPartialRangeReader(DataSegment dataSegment,
PartialLoadSpec wrapper)
+ throws SegmentLoadingException
+ {
+ try {
+ return wrapper.openRangeReader();
+ }
+ catch (IOException e) {
+ throw new SegmentLoadingException(e, "Failed to open range reader for
segment[%s]", dataSegment.getId());
+ }
+ }
+
+ /**
+ * Handle the rangeReader-null path: the backend can't support partial
loads, so the rule can't be honored. Do
+ * static-side reconciliation only; a stale static entry from a prior rule
with a different fingerprint would keep
+ * serving the old selection unless dropped.
+ */
+ private void handleUnsupportedRangeReaderFallback(DataSegment dataSegment,
PartialLoadSpec wrapper)
+ {
+ log.warn(
+ "Backend for segment[%s] does not support range reads; partial-load
rule[fingerprint=%s] cannot be honored, "
+ + "falling back to weak full-load at query time",
+ dataSegment.getId(),
+ wrapper.getFingerprint()
+ );
+ final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(dataSegment.getId());
+ final ReferenceCountingLock lock = lock(dataSegment);
+ synchronized (lock) {
+ try {
+ boolean dropStaleStatic = false;
+ for (StorageLocation location : locations) {
+ if (!location.isReserved(id)) {
+ continue;
+ }
+ final SegmentCacheEntry existing = location.getCacheEntry(id);
+ if (existing instanceof PartialSegmentMetadataCacheEntry
existingPartial
+ && Objects.equals(existingPartial.getFingerprint(),
wrapper.getFingerprint())) {
+ // Same-rule re-issue: preserve the existing entry's info file
across its eventual drop by clearing
+ // the onUnmount hook, matching the fingerprint-match
short-circuit in the main reconciliation branch.
+ existing.setOnUnmount(null);
+ return;
+ }
+ dropStaleStatic = true;
+ }
+ if (dropStaleStatic) {
+ log.info(
+ "Dropping stale static partial entry for segment[%s]: current
rule cannot be applied without range reads",
+ dataSegment.getId()
+ );
+ drop(dataSegment);
Review Comment:
[P2] Range-reader fallback drops live partial entries
This unsupported-range-reader fallback calls `drop()` and then removes every
stale per-segment directory without the `cascadeReleaseWouldDeferCleanup()`
guard used by the main reconciliation path. If a query is still holding the old
static partial metadata or bundle when a changed rule falls back because the
delegate has no range reader, `drop()` can defer old cleanup while
`removeStaleSegmentDirsOnAllLocations()` deletes or reuses the same cache
directory underneath that query. Please apply the same outstanding-reference
precheck here and retry later instead of dropping when cleanup would defer.
--
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]
