feat: Add CONTAINERSIZE segment metadata analysis type

benchmarks/src/test/java/org/apache/druid/benchmark/DruidSchemaInternRowSignatureBenchmark.java

@@ -134,18 +134,13 @@
       return Sequences.simple(
           Lists.transform(
               Lists.newArrayList(segments),
-              (segment) -> new SegmentAnalysis(
-                  segment.toString(),
-                  ImmutableList.of(segment.getInterval()),
-                  columnToAnalysisMap,
-                  40,
-                  40,
-                  null,
-                  null,
-                  null,
-                  null,
-                  false
-              )
+              (segment) -> new SegmentAnalysis.Builder(segment)
+                  .interval(segment.getInterval())
+                  .columns(columnToAnalysisMap)
+                  .size(40)
+                  .numRows(40)
+                  .rollup(false)
+                  .build()
           )
       );
     }

indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java

@@ -126,6 +126,7 @@
 import org.apache.druid.segment.data.CompressionStrategy;
 import org.apache.druid.segment.data.ListIndexed;
 import org.apache.druid.segment.data.RoaringBitmapSerdeFactory;
+import org.apache.druid.segment.file.NoopSegmentFileMapper;
 import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
 import org.apache.druid.segment.indexing.BatchIOConfig;
 import org.apache.druid.segment.indexing.CombinedDataSchema;
@@ -2354,7 +2355,7 @@ private static class TestIndexIO extends IndexIO
                 new ListIndexed<>(segment.getDimensions()),
                 null,
                 columnMap,
-                null
+                NoopSegmentFileMapper.INSTANCE
             )
             {
               @Override

processing/src/main/java/org/apache/druid/java/util/common/io/smoosh/SmooshedFileMapper.java

@@ -27,7 +27,9 @@
 import org.apache.druid.java.util.common.ByteBufferUtils;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.segment.file.SegmentFileMapper;
+import org.apache.druid.segment.file.SegmentFileMetadata;
 
+import javax.annotation.Nullable;
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
@@ -120,6 +122,14 @@ public Set<String> getInternalFilenames()
     return internalFiles.keySet();
   }
 
+  @Nullable
+  @Override
+  public SegmentFileMetadata getSegmentFileMetadata()
+  {
+    // legacy smoosh files have no container/bundle structure to report.
+    return null;
+  }
+
   @Override
   public ByteBuffer mapFile(String name) throws IOException
   {

processing/src/main/java/org/apache/druid/query/metadata/SegmentMetadataQueryQueryToolChest.java

@@ -55,6 +55,7 @@
 import org.apache.druid.query.metadata.metadata.AggregatorMergeStrategy;
 import org.apache.druid.query.metadata.metadata.ColumnAnalysis;
 import org.apache.druid.query.metadata.metadata.SegmentAnalysis;
+import org.apache.druid.query.metadata.metadata.SegmentAnalysis.ContainerAnalysis;
 import org.apache.druid.query.metadata.metadata.SegmentMetadataQuery;
 import org.apache.druid.segment.AggregateProjectionMetadata;
 import org.apache.druid.timeline.LogicalSegment;
@@ -457,35 +458,68 @@ public static SegmentAnalysis mergeAnalyses(
       projections = null;
     }
 
-    return new SegmentAnalysis(
-        mergedId,
-        newIntervals,
-        columns,
-        arg1.getSize() + arg2.getSize(),
-        arg1.getNumRows() + arg2.getNumRows(),
-        aggregators.isEmpty() ? null : aggregators,
-        (projections == null || projections.isEmpty()) ? null : projections,
-        timestampSpec,
-        queryGranularity,
-        rollup
-    );
+    // Merged containers report one total per bundle name rather than a raw list of every underlying segment's
+    // individual containers, since a container has no identity across segments (only bundle names do).
+    final List<ContainerAnalysis> containers = mergeContainers(arg1.getContainers(), arg2.getContainers());
+
+    return new SegmentAnalysis.Builder(mergedId)
+        .intervals(newIntervals)
+        .columns(columns)
+        .size(arg1.getSize() + arg2.getSize())
+        .numRows(arg1.getNumRows() + arg2.getNumRows())
+        .aggregators(aggregators)
+        .projections(projections)
+        .timestampSpec(timestampSpec)
+        .queryGranularity(queryGranularity)
+        .rollup(rollup)
+        .containers(containers)
+        .build();
+  }
+
+  /**
+   * Sums container sizes by bundle name, since a container has no identity across segments. If only one side has
+   * data, it's returned unchanged instead of being run through the per-bundle collapse: there's nothing to merge it
+   * with, and a single segment's own list may already have more than one entry for the same bundle (a bundle
+   * spanning multiple containers).
+   */
+  @Nullable
+  private static List<ContainerAnalysis> mergeContainers(
+      @Nullable List<ContainerAnalysis> containers1,
+      @Nullable List<ContainerAnalysis> containers2
+  )
+  {
+    if (containers1 == null) {
+      return containers2;
+    }
+    if (containers2 == null) {
+      return containers1;
+    }
+    final Map<String, Long> sizeByBundle = new LinkedHashMap<>();
+    for (ContainerAnalysis container : Iterables.concat(containers1, containers2)) {
+      sizeByBundle.merge(container.bundle(), container.size(), Long::sum);
+    }
+    final List<ContainerAnalysis> merged = new ArrayList<>(sizeByBundle.size());
+    for (Map.Entry<String, Long> entry : sizeByBundle.entrySet()) {
+      merged.add(new ContainerAnalysis(entry.getKey(), entry.getValue()));
+    }
+    return merged;
   }
 
   @VisibleForTesting
   public static SegmentAnalysis finalizeAnalysis(SegmentAnalysis analysis)
   {
-    return new SegmentAnalysis(
-        analysis.getId(),
-        analysis.getIntervals() != null ? JodaUtils.condenseIntervals(analysis.getIntervals()) : null,
-        analysis.getColumns(),
-        analysis.getSize(),
-        analysis.getNumRows(),
-        analysis.getAggregators(),
-        analysis.getProjections(),
-        analysis.getTimestampSpec(),
-        analysis.getQueryGranularity(),
-        analysis.isRollup()
-    );
+    return new SegmentAnalysis.Builder(analysis.getId())
+        .intervals(analysis.getIntervals() != null ? JodaUtils.condenseIntervals(analysis.getIntervals()) : null)
+        .columns(analysis.getColumns())
+        .size(analysis.getSize())
+        .numRows(analysis.getNumRows())
+        .aggregators(analysis.getAggregators())
+        .projections(analysis.getProjections())
+        .timestampSpec(analysis.getTimestampSpec())
+        .queryGranularity(analysis.getQueryGranularity())
+        .rollup(analysis.isRollup())
+        .containers(analysis.getContainers())
+        .build();
   }
 
   public SegmentMetadataQueryConfig getConfig()

processing/src/main/java/org/apache/druid/query/metadata/SegmentMetadataQueryRunnerFactory.java

@@ -36,10 +36,13 @@
 import org.apache.druid.query.metadata.metadata.ColumnAnalysis;
 import org.apache.druid.query.metadata.metadata.ColumnIncluderator;
 import org.apache.druid.query.metadata.metadata.SegmentAnalysis;
+import org.apache.druid.query.metadata.metadata.SegmentAnalysis.ContainerAnalysis;
 import org.apache.druid.query.metadata.metadata.SegmentMetadataQuery;
 import org.apache.druid.segment.AggregateProjectionMetadata;
 import org.apache.druid.segment.Metadata;
+import org.apache.druid.segment.QueryableIndex;
 import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.file.SegmentFileContainerMetadata;
 import org.joda.time.Interval;
 
 import javax.annotation.Nullable;
@@ -165,22 +168,33 @@ public Sequence<SegmentAnalysis> run(QueryPlus<SegmentAnalysis> inQ, ResponseCon
           }
         }
 
-        return Sequences.simple(
-            Collections.singletonList(
-                new SegmentAnalysis(
-                    segment.getId().toString(),
-                    retIntervals,
-                    columns,
-                    totalSize,
-                    numRows,
-                    aggregators,
-                    projectionsMap,
-                    timestampSpec,
-                    queryGranularity,
-                    rollup
-                )
-            )
-        );
+        final List<ContainerAnalysis> containers;
+        if (updatedQuery.hasContainerSizes()) {
+          final QueryableIndex index = segment.as(QueryableIndex.class);
+          final List<SegmentFileContainerMetadata> fileContainers = index == null ? null : index.getFileContainers();
+          containers = fileContainers == null
+                       ? null
+                       : fileContainers.stream()
+                                       .map(c -> new ContainerAnalysis(c.getBundle(), c.getSize()))
+                                       .collect(Collectors.toList());
+        } else {
+          containers = null;
+        }
+
+        final SegmentAnalysis analysis = new SegmentAnalysis.Builder(segment.getId().toString())
+            .intervals(retIntervals)
+            .columns(columns)
+            .size(totalSize)
+            .numRows(numRows)
+            .aggregators(aggregators)
+            .projections(projectionsMap)
+            .timestampSpec(timestampSpec)
+            .queryGranularity(queryGranularity)
+            .rollup(rollup)
+            .containers(containers)
+            .build();
+
+        return Sequences.simple(Collections.singletonList(analysis));
       }
     };
   }