feat: wire up partial load rules to segment cache

processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java

@@ -448,7 +448,8 @@ public void ensureAllDownloaded() throws IOException
   }
 
   /**
-   * Download every container belonging to {@code bundleName} in this mapper, each in a single range read (see
+   * Download every container belonging to {@code bundleName}, recursing into external mappers to cover bundles that
+   * span the main file plus one or more externals. Each container is downloaded in a single range read (see
    * {@link #downloadContainer}). No-op for an unknown bundle.
    */
   public void ensureBundleDownloaded(String bundleName) throws IOException
@@ -457,6 +458,32 @@ public void ensureBundleDownloaded(String bundleName) throws IOException
     for (int containerIndex : getContainerIndicesForBundle(bundleName)) {
       downloadContainer(containerIndex);
     }
+    for (PartialSegmentFileMapperV10 external : externalMappers.values()) {
+      external.ensureBundleDownloaded(bundleName);
+    }
+  }
+
+  /**
+   * Whether every container belonging to {@code bundleName} has all its files present in {@link #downloadedFiles},
+   * across the main mapper AND every attached external mapper.
+   * <p>
+   * Returns {@code true} for a bundle name unknown to any mapper (no containers to check).
+   */
+  public boolean isBundleFullyDownloaded(String bundleName)
+  {
+    checkClosed();
+    for (int containerIndex : getContainerIndicesForBundle(bundleName)) {
+      final List<String> fileNames = containerFileNames.get(containerIndex);
+      if (!downloadedFiles.containsAll(fileNames)) {
+        return false;
+      }
+    }
+    for (PartialSegmentFileMapperV10 external : externalMappers.values()) {
+      if (!external.isBundleFullyDownloaded(bundleName)) {
+        return false;
+      }
+    }
+    return true;
   }
 
   /**

processing/src/main/java/org/apache/druid/segment/loading/PartialClusterGroupLoadSpec.java

@@ -25,7 +25,16 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Preconditions;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.segment.projections.ClusteredValueGroupsBaseTableSchema;
+import org.apache.druid.segment.projections.ProjectionMetadata;
+import org.apache.druid.segment.projections.Projections;
+import org.apache.druid.segment.projections.TableClusterGroupSpec;
+import org.apache.druid.timeline.ClusterGroupTuples;
+import org.apache.druid.timeline.DataSegment;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -83,6 +92,64 @@ public List<Integer> getClusterGroupIndices()
     return clusterGroupIndices;
   }
 
+  /**
+   * Resolve each {@code clusterGroupIndex} against the segment's {@link ClusteredValueGroupsBaseTableSchema} to a
+   * {@code __base$<clusteringValueIds>} bundle name via {@link Projections#getClusterGroupBundleName}. The base
+   * projection carries the authoritative group list; each index picks the group at that position and its
+   * {@code clusteringValueIds} disambiguate the bundle in the V10 layout.
+   * <p>
+   * Defensive tripwires fire on any structural inconsistency: empty projections list, non-clustered base projection,
+   * segment's cluster-group tuple count vs. metadata's group count mismatch, or index out of range. These conditions
+   * are unreachable under a healthy writer/reader contract; a throw here indicates a coding bug.
+   * <p>
+   * Returns an empty list when {@code clusterGroupIndices} is empty (the "sibling-empty" case where a matcher
+   * applied to a clustered segment but no configured pattern matched any group).
+   */
+  @Override
+  public List<String> getSelectedBundleNames(DataSegment segment, SegmentFileMetadata metadata)
+  {
+    if (clusterGroupIndices.isEmpty()) {
+      return List.of();
+    }
+    final List<ProjectionMetadata> projections = metadata.getProjections();
+    if (projections == null || projections.isEmpty()) {
+      throw DruidException.defensive(
+          "Cannot resolve cluster-group bundle names for segment[%s]: metadata has no projections",
+          segment.getId()
+      );
+    }
+    if (!(projections.getFirst().getSchema() instanceof ClusteredValueGroupsBaseTableSchema clusteredSummary)) {
+      throw DruidException.defensive(
+          "Cannot resolve cluster-group bundle names for segment[%s]: base projection is not clustered",
+          segment.getId()
+      );
+    }
+    final List<TableClusterGroupSpec> metadataGroups = clusteredSummary.getClusterGroups();
+    final ClusterGroupTuples segmentTuples = segment.getClusterGroups();
+    final int segmentTupleCount = segmentTuples == null ? 0 : segmentTuples.tuples().size();
+    if (segmentTupleCount != metadataGroups.size()) {
+      throw DruidException.defensive(
+          "Cluster-group count mismatch for segment[%s]: DataSegment has [%s] tuples, metadata has [%s] groups",
+          segment.getId(),
+          segmentTupleCount,
+          metadataGroups.size()
+      );
+    }
+    final List<String> bundleNames = new ArrayList<>(clusterGroupIndices.size());
+    for (int idx : clusterGroupIndices) {
+      if (idx < 0 || idx >= metadataGroups.size()) {
+        throw DruidException.defensive(
+            "Cluster-group index [%s] is out of range [0, %s) for segment[%s]",
+            idx,
+            metadataGroups.size(),
+            segment.getId()
+        );
+      }
+      bundleNames.add(Projections.getClusterGroupBundleName(metadataGroups.get(idx).getClusteringValueIds()));
+    }
+    return bundleNames;
+  }
+
   @Override
   public boolean equals(Object o)
   {

processing/src/main/java/org/apache/druid/segment/loading/PartialLoadSpec.java

@@ -24,10 +24,13 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.timeline.DataSegment;
 
 import javax.annotation.Nullable;
 import java.io.File;
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -136,4 +139,17 @@ public SegmentRangeReader openRangeReader() throws IOException
   {
     return materializedDelegate().openRangeReader();
   }
+
+  /**
+   * Returns the cache-layer bundle names that this partial-load spec selects for {@code segment}, resolving the
+   * scheme-specific identifiers carried in the load spec against the segment's parsed {@code metadata}. The
+   * historical's cache layer uses the returned names to acquire rule holds on the matching bundles as part of
+   * {@link SegmentCacheManager#load} for a partial-load rule; bundles not in the returned list are left to the
+   * ordinary on-demand acquire path.
+   * <p>
+   * Returns an empty list when the load spec's selection is empty (the "sibling-empty" case — coordinator matcher
+   * applied but produced no positive selection; the historical still installs a rule hold on the metadata so
+   * bootstrap re-discovers the segment).
+   */
+  public abstract List<String> getSelectedBundleNames(DataSegment segment, SegmentFileMetadata metadata);
 }

processing/src/main/java/org/apache/druid/segment/loading/PartialProjectionLoadSpec.java

@@ -25,11 +25,17 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Preconditions;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.segment.projections.ProjectionMetadata;
+import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.utils.CollectionUtils;
 
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * A {@link PartialLoadSpec} that requests partial loading of a segment's projections. The base class carries the
@@ -84,6 +90,42 @@ public List<String> getProjections()
     return projections;
   }
 
+  /**
+   * Projection names double as bundle names in the V10 partial-segment layout (each projection's containers are
+   * tagged with its name as the bundle prefix), so the load spec selection maps to bundle names verbatim after
+   * validating that each requested name refers to a projection actually present on the segment.
+   * <p>
+   * These are pure defensive tripwires: the coordinator-side matcher derives the wire form from the same segment
+   * metadata this method reads, so a mismatch here indicates a coding bug (matcher/reader drift, writer/reader
+   * contract violation, or serialization corruption).
+   */
+  @Override
+  public List<String> getSelectedBundleNames(DataSegment segment, SegmentFileMetadata metadata)
+  {
+    final List<ProjectionMetadata> segmentProjections = metadata.getProjections();
+    if (segmentProjections == null || segmentProjections.isEmpty()) {
+      throw DruidException.defensive(
+          "Cannot resolve projection bundles for segment[%s]: metadata has no projections",
+          segment.getId()
+      );
+    }
+    final Set<String> known = segmentProjections.stream()
+                                                .map(pm -> pm.getSchema().getName())
+                                                .collect(Collectors.toSet());
+    for (String projection : projections) {
+      if (!known.contains(projection)) {
+        throw DruidException.defensive(
+            "Segment[%s] does not contain projection[%s]; matcher/reader drift or writer/reader contract violation."
+            + " Known projections: %s",
+            segment.getId(),
+            projection,
+            known
+        );
+      }
+    }
+    return projections;
+  }
+
   @Override
   public boolean equals(Object o)
   {