FrankChen021 commented on code in PR #19657:
URL: https://github.com/apache/druid/pull/19657#discussion_r3536282539
##########
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] Preserve stale static partials while refs are active
The range-reader-null fallback drops a stale static partial entry and
immediately deletes stale segment directories without the outstanding-reference
precheck used by the normal reconciliation path. If a query still holds a
metadata or bundle reference, `drop` only removes the entry from the static map
while actual cleanup is deferred; deleting/reusing the directory before those
refs close can break in-flight reads or let deferred cleanup delete files for
the next entry. This path should fail/retry when the old partial would defer
cleanup.
##########
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] Evict weak linked bundles during partial drops
The new drop cascade calls `location.release` for each linked bundle, but
`release` is a no-op for weak/on-demand bundles. A rule-loaded partial segment
can still have non-selected bundles mounted through the weak path; those
bundles hold metadata references for their lifetime, so dropping/reconciling
the static metadata can proceed, delete/recreate the same segment directory,
and leave old weak bundle entries to later evict containers through the old
mapper. Drop/reconcile needs to remove unheld weak linked bundles too, or
refuse to proceed while any linked weak bundle remains active.
--
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]
