feat: wire up partial load rules to segment cache

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

@@ -920,7 +920,7 @@ private static void fetchAndPersistHeader(
   /**
    * Parse metadata from the header file. Throws on any parse failure (corrupt JSON, truncated header, etc.).
    */
-  private static SegmentFileMetadataReader.Result parseHeaderFile(
+  public static SegmentFileMetadataReader.Result parseHeaderFile(
       File headerFile,
       ObjectMapper jsonMapper
   ) throws IOException

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,62 @@ public List<Integer> getClusterGroupIndices()
     return clusterGroupIndices;
   }
 
+  /**
+   * Translates each selected {@code clusterGroupIndex} to the bundle name of the corresponding on-disk cluster
+   * group, by finding the {@link TableClusterGroupSpec} at that position in the parsed base-table schema and
+   * formatting its {@link TableClusterGroupSpec#getClusteringValueIds()} via
+   * {@link Projections#getClusterGroupBundleName(List)}.
+   * <p>
+   * Positional alignment between {@link DataSegment#getClusterGroups()} tuples and the metadata's
+   * {@link ClusteredValueGroupsBaseTableSchema#getClusterGroups()} list is invariant of the writer: both are populated
+   * from the same source array at segment-build time. A size-mismatch indicates a writer/reader contract violation
+   * and trips a defensive throw.
+   */
+  @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,15 @@ 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 {@code metadata}. The historical's cache
+   * layer uses the returned names to eagerly mount and pin the matching bundles as static entries; other bundles
+   * remain weak and on-demand.
+   * <p>
+   * Returns an empty list when the load spec carries an empty selection, where only the metadata header is sticky and
+   * no bundle is pinned.
+   */
+  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,38 @@ 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.
+   */
+  @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)
   {

processing/src/test/java/org/apache/druid/segment/loading/PartialClusterGroupLoadSpecTest.java

@@ -27,7 +27,26 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.google.common.collect.ImmutableMap;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.segment.projections.ClusteredValueGroupsBaseTableSchema;
+import org.apache.druid.segment.projections.ClusteringDictionaries;
+import org.apache.druid.segment.projections.ProjectionMetadata;
+import org.apache.druid.segment.projections.ProjectionSchema;
+import org.apache.druid.segment.projections.Projections;
+import org.apache.druid.segment.projections.TableClusterGroupSpec;
+import org.apache.druid.segment.projections.TableProjectionSchema;
+import org.apache.druid.timeline.ClusterGroupTuples;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -163,6 +182,193 @@ void testRejectsNullFingerprint()
     );
   }
 
+  @Test
+  void testGetSelectedBundleNamesResolvesIndicesToBundleNames()
+  {
+    // Three cluster groups in the metadata; pick groups 0 and 2.
+    final SegmentFileMetadata metadata = clusteredMetadata(
+        List.of(
+            new TableClusterGroupSpec(List.of(0), 10),
+            new TableClusterGroupSpec(List.of(1), 20),
+            new TableClusterGroupSpec(List.of(2), 30)
+        )
+    );
+    final DataSegment segment = clusteredSegment(
+        List.of(List.of("acme"), List.of("globex"), List.of("initech"))
+    );
+    final PartialClusterGroupLoadSpec spec = new PartialClusterGroupLoadSpec(
+        DELEGATE,
+        List.of(0, 2),
+        FINGERPRINT,
+        jsonMapper
+    );
+    Assertions.assertEquals(
+        List.of(
+            Projections.getClusterGroupBundleName(List.of(0)),
+            Projections.getClusterGroupBundleName(List.of(2))
+        ),
+        spec.getSelectedBundleNames(segment, metadata)
+    );
+  }
+
+  @Test
+  void testGetSelectedBundleNamesEmptyIndicesReturnsEmpty()
+  {
+    // Sibling-empty case: matcher applied but no positive content. Caller pins only the metadata.
+    final SegmentFileMetadata metadata = clusteredMetadata(
+        List.of(new TableClusterGroupSpec(List.of(0), 1))
+    );
+    final DataSegment segment = clusteredSegment(List.of(List.of("acme")));
+    final PartialClusterGroupLoadSpec spec = new PartialClusterGroupLoadSpec(
+        DELEGATE,
+        List.of(),
+        FINGERPRINT,
+        jsonMapper
+    );
+    Assertions.assertEquals(List.of(), spec.getSelectedBundleNames(segment, metadata));
+  }
+
+  @Test
+  void testGetSelectedBundleNamesOutOfRangeIndexThrows()
+  {
+    final SegmentFileMetadata metadata = clusteredMetadata(
+        List.of(new TableClusterGroupSpec(List.of(0), 1))
+    );
+    final DataSegment segment = clusteredSegment(List.of(List.of("acme")));
+    final PartialClusterGroupLoadSpec spec = new PartialClusterGroupLoadSpec(
+        DELEGATE,
+        List.of(5),
+        FINGERPRINT,
+        jsonMapper
+    );
+    final DruidException thrown = Assertions.assertThrows(
+        DruidException.class,
+        () -> spec.getSelectedBundleNames(segment, metadata)
+    );
+    Assertions.assertTrue(
+        thrown.getMessage().contains("Cluster-group index [5] is out of range"),
+        "unexpected message: " + thrown.getMessage()
+    );
+  }
+
+  @Test
+  void testGetSelectedBundleNamesSizeMismatchThrows()
+  {
+    // Metadata has 2 groups, segment has 1 tuple — writer/reader contract violation; tripwire fires.
+    final SegmentFileMetadata metadata = clusteredMetadata(
+        List.of(
+            new TableClusterGroupSpec(List.of(0), 1),
+            new TableClusterGroupSpec(List.of(1), 1)
+        )
+    );
+    final DataSegment segment = clusteredSegment(List.of(List.of("acme")));
+    final PartialClusterGroupLoadSpec spec = new PartialClusterGroupLoadSpec(
+        DELEGATE,
+        List.of(0),
+        FINGERPRINT,
+        jsonMapper
+    );
+    final DruidException thrown = Assertions.assertThrows(
+        DruidException.class,
+        () -> spec.getSelectedBundleNames(segment, metadata)
+    );
+    Assertions.assertTrue(
+        thrown.getMessage().contains("Cluster-group count mismatch"),
+        "unexpected message: " + thrown.getMessage()
+    );
+  }
+
+  @Test
+  void testGetSelectedBundleNamesNonClusteredBaseThrows()
+  {
+    // Base projection is a regular (non-clustered) table — partial cluster-group load is nonsensical.
+    final ProjectionSchema baseSchema = new TableProjectionSchema(
+        VirtualColumns.EMPTY,
+        List.of(ColumnHolder.TIME_COLUMN_NAME, "tenant"),
+        null,
+        List.of(OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME))
+    );
+    final SegmentFileMetadata metadata = new SegmentFileMetadata(
+        List.of(),
+        Map.of(),
+        null,
+        null,
+        List.of(new ProjectionMetadata(1, baseSchema)),
+        null
+    );
+    final DataSegment segment = clusteredSegment(List.of(List.of("acme")));
+    final PartialClusterGroupLoadSpec spec = new PartialClusterGroupLoadSpec(
+        DELEGATE,
+        List.of(0),
+        FINGERPRINT,
+        jsonMapper
+    );
+    final DruidException thrown = Assertions.assertThrows(
+        DruidException.class,
+        () -> spec.getSelectedBundleNames(segment, metadata)
+    );
+    Assertions.assertTrue(
+        thrown.getMessage().contains("base projection is not clustered"),
+        "unexpected message: " + thrown.getMessage()
+    );
+  }
+
+  @Test
+  void testGetSelectedBundleNamesNoProjectionsThrows()
+  {
+    final SegmentFileMetadata metadata = new SegmentFileMetadata(List.of(), Map.of(), null, null, null, null);
+    final DataSegment segment = clusteredSegment(List.of(List.of("acme")));
+    final PartialClusterGroupLoadSpec spec = new PartialClusterGroupLoadSpec(
+        DELEGATE,
+        List.of(0),
+        FINGERPRINT,
+        jsonMapper
+    );
+    final DruidException thrown = Assertions.assertThrows(
+        DruidException.class,
+        () -> spec.getSelectedBundleNames(segment, metadata)
+    );
+    Assertions.assertTrue(
+        thrown.getMessage().contains("metadata has no projections"),
+        "unexpected message: " + thrown.getMessage()
+    );
+  }
+
+  private static final RowSignature CLUSTERING_TENANT = RowSignature.builder()
+                                                                    .add("tenant", ColumnType.STRING)
+                                                                    .build();
+
+  private static SegmentFileMetadata clusteredMetadata(List<TableClusterGroupSpec> groups)
+  {
+    final ClusteredValueGroupsBaseTableSchema baseSchema = new ClusteredValueGroupsBaseTableSchema(
+        VirtualColumns.EMPTY,
+        List.of(ColumnHolder.TIME_COLUMN_NAME, "tenant", "metric"),
+        List.of(OrderBy.ascending("tenant"), OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME)),
+        CLUSTERING_TENANT,
+        null,
+        new ClusteringDictionaries(List.of("acme", "globex", "initech"), null, null, null),
+        groups
+    );
+    return new SegmentFileMetadata(
+        List.of(),
+        Map.of(),
+        null,
+        null,
+        List.of(new ProjectionMetadata(groups.stream().mapToInt(TableClusterGroupSpec::getNumRows).sum(), baseSchema)),
+        null
+    );
+  }
+
+  private static DataSegment clusteredSegment(List<List<Object>> tuples)
+  {
+    return DataSegment.builder(
+                          SegmentId.of("ds", Intervals.ETERNITY, "v1", new NumberedShardSpec(0, 1))
+                      )
+                      .size(0)
+                      .clusterGroups(new ClusterGroupTuples(CLUSTERING_TENANT, tuples))
+                      .build();
+  }
+
   /**
    * Stub LoadSpec used to verify delegation. Uses the same JSON "type"=="stub" key as the test {@link #DELEGATE}.
    */