diff --git a/processing/src/main/java/org/apache/druid/java/util/common/FileUtils.java b/processing/src/main/java/org/apache/druid/java/util/common/FileUtils.java index de66027615e6..f852786cc8b2 100644 --- a/processing/src/main/java/org/apache/druid/java/util/common/FileUtils.java +++ b/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. + *

+ * 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". diff --git a/processing/src/main/java/org/apache/druid/segment/DictionaryEncodedColumnMerger.java b/processing/src/main/java/org/apache/druid/segment/DictionaryEncodedColumnMerger.java index 8ea88b580f67..1cd33a857623 100644 --- a/processing/src/main/java/org/apache/druid/segment/DictionaryEncodedColumnMerger.java +++ b/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/", 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); } } } diff --git a/processing/src/test/java/org/apache/druid/java/util/common/FileUtilsTest.java b/processing/src/test/java/org/apache/druid/java/util/common/FileUtilsTest.java index b410c0b64196..d242d1fd211f 100644 --- a/processing/src/test/java/org/apache/druid/java/util/common/FileUtilsTest.java +++ b/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 { diff --git a/processing/src/test/java/org/apache/druid/segment/IndexMergerV10MinMaxTimeTest.java b/processing/src/test/java/org/apache/druid/segment/IndexMergerV10MinMaxTimeTest.java index cddd50e47c0c..ddae69443a6b 100644 --- a/processing/src/test/java/org/apache/druid/segment/IndexMergerV10MinMaxTimeTest.java +++ b/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/). + // 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 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 rows, DimensionsSpec dimensionsSpec) + { + return buildV10Segment(rows, dimensionsSpec, Collections.emptyList()); + } + + private File buildV10Segment( + List rows, + DimensionsSpec dimensionsSpec, + List projections + ) { final long minTs = rows.stream().mapToLong(InputRow::getTimestampFromEpoch).min().orElseThrow(); return IndexBuilder.create() @@ -123,6 +186,7 @@ private File buildV10Segment(List rows, DimensionsSpec dimensionsSpec) .withDimensionsSpec(dimensionsSpec) .withRollup(false) .withMinTimestamp(minTs) + .withProjections(projections) .build() ) .rows(rows) diff --git a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java index f5d2452d9c38..2c6e4505a5a8 100644 --- a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java +++ b/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); - } - } } /**