minor: add fileGroup to V10 segment SegmentFileContainerMetadata, switch default metadata compression to zstd

embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryVirtualStorageTest.java

@@ -66,8 +66,9 @@
  */
 class QueryVirtualStorageTest extends EmbeddedClusterTestBase
 {
-  // size of wiki segments, adjust this if segment size changes for some reason
-  private static final long SIZE_BYTES = 3777834L;
+  // size of wiki segments (size here is size with uncompressed metadata as an upper bound since the zstd default
+  // appears to make different sizes on different platforms) adjust this if segment size changes for some reason
+  private static final long SIZE_BYTES = 3778338L;
   private static final long CACHE_SIZE = HumanReadableBytes.parse("1MiB");
   private static final long MAX_SIZE = HumanReadableBytes.parse("100MiB");
 
@@ -305,11 +306,12 @@ void testQueryTooMuchDataButWithDart()
   @Test
   void testQuerySysTables()
   {
-    String query = "SELECT curr_size, max_size, storage_size FROM sys.servers WHERE tier IS NOT NULL AND server_type = 'historical'";
-    Assertions.assertEquals(
-        StringUtils.format("%s,%s,%s", SIZE_BYTES, MAX_SIZE, CACHE_SIZE),
-        cluster.callApi().runSql(query)
-    );
+    final String query = "SELECT curr_size, max_size, storage_size FROM sys.servers WHERE tier IS NOT NULL AND server_type = 'historical'";
+    final String resultString = cluster.callApi().runSql(query);
+    final String[] split = resultString.split(",");
+    Assertions.assertTrue(Long.parseLong(split[0]) <= SIZE_BYTES);
+    Assertions.assertEquals(MAX_SIZE, Long.parseLong(split[1]));
+    Assertions.assertEquals(CACHE_SIZE, Long.parseLong(split[2]));
   }
 
 

processing/src/main/java/org/apache/druid/segment/IndexSpec.java

@@ -29,6 +29,7 @@
 import org.apache.druid.segment.data.BitmapSerdeFactory;
 import org.apache.druid.segment.data.CompressionFactory;
 import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.file.SegmentFileBuilderV10;
 import org.apache.druid.segment.loading.SegmentizerFactory;
 import org.apache.druid.segment.nested.NestedCommonFormatColumnFormatSpec;
 
@@ -249,7 +250,7 @@ public IndexSpec getEffectiveSpec()
     } else if (defaultSpec.metadataCompression != null) {
       bob.withMetadataCompression(defaultSpec.metadataCompression);
     } else {
-      bob.withMetadataCompression(CompressionStrategy.NONE);
+      bob.withMetadataCompression(SegmentFileBuilderV10.DEFAULT_METADATA_COMPRESSION);
     }
 
     if (dimensionCompression != null) {

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

@@ -80,9 +80,14 @@ public class SegmentFileBuilderV10 implements SegmentFileBuilder
 {
   private static final Logger LOG = new Logger(SegmentFileBuilderV10.class);
 
+  /**
+   * Default compression for the V10 metadata header
+   */
+  public static final CompressionStrategy DEFAULT_METADATA_COMPRESSION = CompressionStrategy.ZSTD;
+
   public static SegmentFileBuilderV10 create(ObjectMapper jsonMapper, File baseDir)
   {
-    return create(jsonMapper, baseDir, CompressionStrategy.NONE);
+    return create(jsonMapper, baseDir, DEFAULT_METADATA_COMPRESSION);
   }
 
   public static SegmentFileBuilderV10 create(ObjectMapper jsonMapper, File baseDir, CompressionStrategy metaCompression)
@@ -184,6 +189,7 @@ public SegmentFileChannel addWithChannel(final String name, final long size) thr
     if (internalFiles.containsKey(name)) {
       throw new IAE("Cannot add files of the same name, already have [%s]", name);
     }
+    ensureNameMatchesActiveGroup(name);
     if (size > maxContainerSize) {
       throw DruidException.forPersona(DruidException.Persona.ADMIN)
                           .ofCategory(DruidException.Category.RUNTIME_FAILURE)
@@ -285,9 +291,29 @@ public SegmentFileBuilder getExternalBuilder(String externalFile)
   @Override
   public void addColumn(String name, ColumnDescriptor columnDescriptor)
   {
+    ensureNameMatchesActiveGroup(name);
     this.columns.put(name, columnDescriptor);
   }
 
+  /**
+   * If a file group is currently active (set by the most recent {@link #startFileGroup} call), enforce that names of
+   * files and columns added under it are prefixed by {@code groupName + "/"}. Prevents silent collisions where two
+   * groups write a file/column of the same bare name and the second silently overwrites the first in the metadata
+   * maps. Existing production callers (e.g. {@code IndexMergerV10} via
+   * {@code Projections.getProjectionSegmentInternalFileName}) already construct prefixed names, so this is a no-op
+   * for them; it catches new writers that forget the convention.
+   */
+  private void ensureNameMatchesActiveGroup(String name)
+  {
+    if (currentFileGroup != null && !name.startsWith(currentFileGroup + "/")) {
+      throw DruidException.defensive(
+          "Name[%s] must start with the active file group prefix[%s/]",
+          name,
+          currentFileGroup
+      );
+    }
+  }
+
   /**
    * Declare the file group that subsequent writes belong to. Writes are routed into a container tagged with the
    * declared group; a new container is rolled when the group changes or the incoming file won't fit. A group whose
@@ -438,7 +464,7 @@ private List<SegmentFileContainerMetadata> buildContainerMetadata()
     long offset = 0;
     for (ContainerWriter container : containers) {
       final long length = container.file.length();
-      result.add(new SegmentFileContainerMetadata(offset, length));
+      result.add(new SegmentFileContainerMetadata(offset, length, container.group));
       offset += length;
     }
     return result;

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

@@ -20,25 +20,40 @@
 package org.apache.druid.segment.file;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import javax.annotation.Nullable;
+import java.util.Objects;
+
 /**
  * Starting offset and size of a 'container' stored in a V10 segment file; think the V10 equivalent of V9's external
  * 'smoosh' files, e.g. 00000.smoosh.
+ * <p>
+ * Each container holds internal files belonging to at most one named file group, as declared at write time via
+ * {@link SegmentFileBuilder#startFileGroup}. The {@link #fileGroup} field records that name so readers can attribute
+ * a container to its group without parsing internal-file names. The field is {@code null} for containers written
+ * without a {@code startFileGroup} call (or with {@code startFileGroup(null)}), and for containers from segments
+ * produced by writers that pre-date this field; null serializes as a Jackson-omitted property so old segments
+ * round-trip unchanged.
  */
 public class SegmentFileContainerMetadata
 {
   private final long startOffset;
   private final long size;
+  @Nullable
+  private final String fileGroup;
 
   @JsonCreator
   public SegmentFileContainerMetadata(
       @JsonProperty("startOffset") long startOffset,
-      @JsonProperty("size") long size
+      @JsonProperty("size") long size,
+      @JsonProperty("fileGroup") @Nullable String fileGroup
   )
   {
     this.startOffset = startOffset;
     this.size = size;
+    this.fileGroup = fileGroup;
   }
 
   @JsonProperty
@@ -52,4 +67,43 @@ public long getSize()
   {
     return size;
   }
+
+  @JsonProperty
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  @Nullable
+  public String getFileGroup()
+  {
+    return fileGroup;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    SegmentFileContainerMetadata that = (SegmentFileContainerMetadata) o;
+    return startOffset == that.startOffset
+           && size == that.size
+           && Objects.equals(fileGroup, that.fileGroup);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(startOffset, size, fileGroup);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "SegmentFileContainerMetadata{"
+           + "startOffset=" + startOffset
+           + ", size=" + size
+           + ", fileGroup=" + fileGroup
+           + '}';
+  }
 }