fix: projections on base table columns result in failed indexing when using unzipped segments

processing/src/main/java/org/apache/druid/java/util/common/FileUtils.java

@@ -499,6 +499,39 @@ public static void deleteDirectory(final File directory) throws IOException
     org.apache.commons.io.FileUtils.deleteDirectory(directory);
   }
 
+  /**
+   * Deletes {@code directory} (recursively, like {@link #deleteDirectory(File)}), then walks up deleting each
+   * now-empty ancestor directory, stopping at the first non-empty ancestor or when {@code stopAt} is reached,
+   * whichever comes first. {@code stopAt} itself is never deleted, so callers can pass a base directory that must
+   * survive.
+   * <p>
+   * Because an ancestor is removed only once it is empty, this is safe when intermediate directories are shared by
+   * several sibling leaves: the shared ancestor is deleted only after its last child is gone. It is not safe for
+   * concurrent deletion of overlapping paths under the same {@code stopAt}.
+   */
+  public static void deleteDirectoryAndEmptyAncestors(final File directory, final File stopAt) throws IOException
+  {
+    if (directory == null || directory.equals(stopAt)) {
+      return;
+    }
+    deleteDirectory(directory);
+    // Walk up removing now-empty ancestors.
+    File parent = directory.getParentFile();
+    while (parent != null && !parent.equals(stopAt)) {
+      final String[] children = parent.list();
+      if (children == null || children.length > 0) {
+        // Non-empty, or contents could not be listed: leave it in place and stop climbing.
+        break;
+      }
+      // delete() removes only an empty directory. If a concurrent write landed a child between the list() above and
+      // here, the delete fails, and we stop rather than removing a directory that just gained content.
+      if (!parent.delete()) {
+        break;
+      }
+      parent = parent.getParentFile();
+    }
+  }
+
   /**
    * Hard-link "src" as "dest", if possible. If not possible -- perhaps they are on separate filesystems -- then
    * copy "src" to "dest".

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

@@ -161,7 +161,7 @@ public void markAsParent()
     catch (IOException e) {
       throw new RuntimeException(e);
     }
-    persistedIdConversions = closer.register(new PersistedIdConversions(tmpOutputFilesDir));
+    persistedIdConversions = closer.register(new PersistedIdConversions(tmpOutputFilesDir, segmentBaseDir));
   }
 
   @Override
@@ -789,11 +789,13 @@ public void writeTo(WritableByteChannel channel, SegmentFileBuilder fileBuilder)
   protected static class PersistedIdConversions implements Closeable
   {
     private final File tempDir;
+    private final File baseDir;
     private final Closer closer;
 
-    protected PersistedIdConversions(File tempDir)
+    protected PersistedIdConversions(File tempDir, File baseDir)
     {
       this.tempDir = tempDir;
+      this.baseDir = baseDir;
       this.closer = Closer.create();
     }
 
@@ -813,7 +815,10 @@ public void close() throws IOException
         closer.close();
       }
       finally {
-        FileUtils.deleteDirectory(tempDir);
+        // tempDir may be nested under baseDir (its name can carry a bundle prefix such as "__base/<col>", so mkdirp
+        // created intermediate directories). Delete tempDir and any now-empty intermediate directories up to baseDir
+        // so no empty scratch directory is left behind in the finalized segment directory.
+        FileUtils.deleteDirectoryAndEmptyAncestors(tempDir, baseDir);
       }
     }
   }

processing/src/test/java/org/apache/druid/java/util/common/FileUtilsTest.java

@@ -53,6 +53,61 @@ public void testMap() throws IOException
     Assertions.assertEquals(buffersMemoryBefore, buffersMemoryAfter);
   }
 
+  @Test
+  public void testDeleteDirectoryAndEmptyAncestorsRemovesEmptyIntermediateDirs() throws IOException
+  {
+    // base/mid/leaf, where 'leaf' is the scratch dir and 'mid' is an intermediate dir mkdirp created along the way.
+    final File mid = new File(temporaryFolder, "mid");
+    final File leaf = new File(mid, "leaf");
+    FileUtils.mkdirp(leaf);
+
+    FileUtils.deleteDirectoryAndEmptyAncestors(leaf, temporaryFolder);
+
+    Assertions.assertFalse(leaf.exists(), "leaf should be deleted");
+    Assertions.assertFalse(mid.exists(), "empty intermediate dir should be deleted");
+    Assertions.assertTrue(temporaryFolder.exists(), "base (stopAt) must survive");
+  }
+
+  @Test
+  public void testDeleteDirectoryAndEmptyAncestorsStopsAtNonEmptyAncestor() throws IOException
+  {
+    // Shared intermediate dir with two sibling leaves; deleting one leaf must leave the shared parent (and sibling).
+    final File shared = new File(temporaryFolder, "shared");
+    final File leafA = new File(shared, "leafA");
+    final File leafB = new File(shared, "leafB");
+    FileUtils.mkdirp(leafA);
+    FileUtils.mkdirp(leafB);
+
+    FileUtils.deleteDirectoryAndEmptyAncestors(leafA, temporaryFolder);
+
+    Assertions.assertFalse(leafA.exists(), "deleted leaf should be gone");
+    Assertions.assertTrue(leafB.exists(), "sibling leaf must survive");
+    Assertions.assertTrue(shared.exists(), "non-empty shared ancestor must survive");
+
+    // Deleting the last sibling then reclaims the now-empty shared ancestor, stopping at base.
+    FileUtils.deleteDirectoryAndEmptyAncestors(leafB, temporaryFolder);
+    Assertions.assertFalse(shared.exists(), "shared ancestor should be reclaimed once empty");
+    Assertions.assertTrue(temporaryFolder.exists(), "base (stopAt) must survive");
+  }
+
+  @Test
+  public void testDeleteDirectoryAndEmptyAncestorsDeletesNonEmptyLeafButNeverStopAt() throws IOException
+  {
+    // The leaf itself is deleted recursively even when non-empty; a leaf directly under stopAt leaves stopAt intact.
+    final File leaf = new File(temporaryFolder, "leaf");
+    FileUtils.mkdirp(leaf);
+    Assertions.assertTrue(new File(leaf, "buffer").createNewFile());
+
+    FileUtils.deleteDirectoryAndEmptyAncestors(leaf, temporaryFolder);
+
+    Assertions.assertFalse(leaf.exists(), "non-empty leaf should be deleted recursively");
+    Assertions.assertTrue(temporaryFolder.exists(), "base (stopAt) must survive");
+
+    // Passing stopAt itself is a no-op.
+    FileUtils.deleteDirectoryAndEmptyAncestors(temporaryFolder, temporaryFolder);
+    Assertions.assertTrue(temporaryFolder.exists(), "stopAt must never be deleted");
+  }
+
   @Test
   public void testMapFileTooLarge() throws IOException
   {

processing/src/test/java/org/apache/druid/segment/IndexMergerV10MinMaxTimeTest.java

@@ -21,10 +21,12 @@
 
 import org.apache.druid.data.input.InputRow;
 import org.apache.druid.data.input.ListBasedInputRow;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
 import org.apache.druid.data.input.impl.DimensionsSpec;
 import org.apache.druid.data.input.impl.LongDimensionSchema;
 import org.apache.druid.data.input.impl.StringDimensionSchema;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
 import org.apache.druid.segment.file.SegmentFileMapperV10;
@@ -40,6 +42,7 @@
 
 import java.io.File;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 class IndexMergerV10MinMaxTimeTest extends InitializedNullHandlingTest
@@ -111,7 +114,67 @@ void testMinMaxTimePersistedForNonTimeSortedSegment() throws Exception
     assertBaseProjectionMinMaxTime(segmentDir, base.getMillis(), base.plusMinutes(20).getMillis());
   }
 
+  @Test
+  void testProjectionOnBaseDimensionLeavesNoScratchDirsInOutput() throws Exception
+  {
+    // A projection grouped on a base-table dimension marks that dimension's base merger as a projection 'parent',
+    // which persists id-conversion buffers under a scratch dir named from the merger's V10 output name (__base/<dim>).
+    // The embedded '/' makes the scratch dir path nest a 'tmp___base' directory inside the segment dir; if the leaf
+    // is cleaned up but the intermediate 'tmp___base' is left behind, the segment dir is no longer flat and the
+    // range-readable (no-zip) deep-storage push path rejects it as an unexpected subdirectory. The merged segment
+    // directory must contain no such scratch directories.
+    final DateTime base = DateTimes.of("2025-01-01");
+    final RowSignature signature = RowSignature.builder()
+                                               .add("dim", ColumnType.STRING)
+                                               .add("metric", ColumnType.LONG)
+                                               .build();
+    final List<InputRow> rows = Arrays.asList(
+        new ListBasedInputRow(signature, base, signature.getColumnNames(), Arrays.asList("a", 1L)),
+        new ListBasedInputRow(signature, base.plusMinutes(5), signature.getColumnNames(), Arrays.asList("b", 2L)),
+        new ListBasedInputRow(signature, base.plusMinutes(10), signature.getColumnNames(), Arrays.asList("a", 3L))
+    );
+
+    final File segmentDir = buildV10Segment(
+        rows,
+        DimensionsSpec.builder()
+                      .setDimensions(
+                          List.of(
+                              new StringDimensionSchema("dim"),
+                              new LongDimensionSchema("metric")
+                          )
+                      )
+                      .build(),
+        List.of(
+            AggregateProjectionSpec.builder("dim_metric_sum")
+                                   .groupingColumns(new StringDimensionSchema("dim"))
+                                   .aggregators(new LongSumAggregatorFactory("sum_metric", "metric"))
+                                   .build()
+        )
+    );
+
+    final File[] scratchDirs = segmentDir.listFiles(
+        f -> f.isDirectory() && f.getName().startsWith("tmp_")
+    );
+    Assertions.assertTrue(
+        scratchDirs == null || scratchDirs.length == 0,
+        () -> {
+          Assertions.assertNotNull(scratchDirs);
+          return "Merged segment dir must be flat, but found leftover merger scratch dir(s): "
+                   + Arrays.stream(scratchDirs).map(File::getName).toList();
+        }
+    );
+  }
+
   private File buildV10Segment(List<InputRow> rows, DimensionsSpec dimensionsSpec)
+  {
+    return buildV10Segment(rows, dimensionsSpec, Collections.emptyList());
+  }
+
+  private File buildV10Segment(
+      List<InputRow> rows,
+      DimensionsSpec dimensionsSpec,
+      List<AggregateProjectionSpec> projections
+  )
   {
     final long minTs = rows.stream().mapToLong(InputRow::getTimestampFromEpoch).min().orElseThrow();
     return IndexBuilder.create()
@@ -123,6 +186,7 @@ private File buildV10Segment(List<InputRow> rows, DimensionsSpec dimensionsSpec)
                                                  .withDimensionsSpec(dimensionsSpec)
                                                  .withRollup(false)
                                                  .withMinTimestamp(minTs)
+                                                 .withProjections(projections)
                                                  .build()
                        )
                        .rows(rows)

server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java

@@ -1338,25 +1338,13 @@ private static void atomicMoveAndDeleteCacheEntryDirectory(final File path)
    */
   private static void cleanupLegacyCacheLocation(final File baseFile, final File cacheFile)
   {
-    if (cacheFile.equals(baseFile)) {
-      return;
-    }
-
     try {
       log.info("Deleting migrated segment directory[%s]", cacheFile);
-      FileUtils.deleteDirectory(cacheFile);
+      FileUtils.deleteDirectoryAndEmptyAncestors(cacheFile, baseFile);
     }
     catch (Exception e) {
       log.warn(e, "Unable to remove directory[%s]", cacheFile);
     }
-
-    File parent = cacheFile.getParentFile();
-    if (parent != null) {
-      File[] children = parent.listFiles();
-      if (children == null || children.length == 0) {
-        cleanupLegacyCacheLocation(baseFile, parent);
-      }
-    }
   }
 
   /**