cleanup temp files for nested column serializer

processing/src/main/java/org/apache/druid/segment/nested/DictionaryIdLookup.java

@@ -58,21 +58,25 @@ public final class DictionaryIdLookup implements Closeable
   private final String name;
   @Nullable
   private final DictionaryWriter<String> stringDictionaryWriter;
+  private Path stringDictionaryFile = null;
   private SmooshedFileMapper stringBufferMapper = null;
   private Indexed<ByteBuffer> stringDictionary = null;
 
   @Nullable
   private final DictionaryWriter<Long> longDictionaryWriter;
+  private Path longDictionaryFile = null;
   private MappedByteBuffer longBuffer = null;
   private FixedIndexed<Long> longDictionary = null;
 
   @Nullable
   private final DictionaryWriter<Double> doubleDictionaryWriter;
+  private Path doubleDictionaryFile = null;
   MappedByteBuffer doubleBuffer = null;
   FixedIndexed<Double> doubleDictionary = null;
 
   @Nullable
   private final DictionaryWriter<int[]> arrayDictionaryWriter;
+  private Path arrayDictionaryFile = null;
   private MappedByteBuffer arrayBuffer = null;
   private FrontCodedIntArrayIndexed arrayDictionary = null;
 
@@ -99,15 +103,20 @@ public int lookupString(@Nullable String value)
       // in the future, we should transition them to using this approach as well (or build a combination smoosher and
       // mapper so that we can have a mutable smoosh)
       File stringSmoosh = FileUtils.createTempDir(StringUtils.urlEncode(name) + "__stringTempSmoosh");
+      stringSmoosh.deleteOnExit();
+      stringDictionaryFile = stringSmoosh.toPath();
       final String fileName = NestedCommonFormatColumnSerializer.getInternalFileName(
           name,
           NestedCommonFormatColumnSerializer.STRING_DICTIONARY_FILE_NAME
       );
-      final FileSmoosher smoosher = new FileSmoosher(stringSmoosh);
-      try (final SmooshedWriter writer = smoosher.addWithSmooshedWriter(
-          fileName,
-          stringDictionaryWriter.getSerializedSize()
-      )) {
+
+      try (
+          final FileSmoosher smoosher = new FileSmoosher(stringSmoosh);
+          final SmooshedWriter writer = smoosher.addWithSmooshedWriter(
+              fileName,
+              stringDictionaryWriter.getSerializedSize()
+          )
+      ) {
         stringDictionaryWriter.writeTo(writer, smoosher);
         writer.close();
         smoosher.close();
@@ -134,11 +143,12 @@ public int lookupString(@Nullable String value)
   public int lookupLong(@Nullable Long value)
   {
     if (longDictionary == null) {
-      final Path longFile = makeTempFile(name + NestedCommonFormatColumnSerializer.LONG_DICTIONARY_FILE_NAME);
-      longBuffer = mapWriter(longFile, longDictionaryWriter);
+      longDictionaryFile = makeTempFile(name + NestedCommonFormatColumnSerializer.LONG_DICTIONARY_FILE_NAME);
+      longBuffer = mapWriter(longDictionaryFile, longDictionaryWriter);
       longDictionary = FixedIndexed.read(longBuffer, TypeStrategies.LONG, ByteOrder.nativeOrder(), Long.BYTES).get();
       // reset position
       longBuffer.position(0);
+      longDictionaryFile.toFile().deleteOnExit();
     }
     final int index = longDictionary.indexOf(value);
     if (index < 0) {
@@ -150,8 +160,8 @@ public int lookupLong(@Nullable Long value)
   public int lookupDouble(@Nullable Double value)
   {
     if (doubleDictionary == null) {
-      final Path doubleFile = makeTempFile(name + NestedCommonFormatColumnSerializer.DOUBLE_DICTIONARY_FILE_NAME);
-      doubleBuffer = mapWriter(doubleFile, doubleDictionaryWriter);
+      doubleDictionaryFile = makeTempFile(name + NestedCommonFormatColumnSerializer.DOUBLE_DICTIONARY_FILE_NAME);
+      doubleBuffer = mapWriter(doubleDictionaryFile, doubleDictionaryWriter);
       doubleDictionary = FixedIndexed.read(
           doubleBuffer,
           TypeStrategies.DOUBLE,
@@ -160,6 +170,7 @@ public int lookupDouble(@Nullable Double value)
       ).get();
       // reset position
       doubleBuffer.position(0);
+      doubleDictionaryFile.toFile().deleteOnExit();
     }
     final int index = doubleDictionary.indexOf(value);
     if (index < 0) {
@@ -171,11 +182,13 @@ public int lookupDouble(@Nullable Double value)
   public int lookupArray(@Nullable int[] value)
   {
     if (arrayDictionary == null) {
-      final Path arrayFile = makeTempFile(name + NestedCommonFormatColumnSerializer.ARRAY_DICTIONARY_FILE_NAME);
-      arrayBuffer = mapWriter(arrayFile, arrayDictionaryWriter);
+      arrayDictionaryFile = makeTempFile(name + NestedCommonFormatColumnSerializer.ARRAY_DICTIONARY_FILE_NAME);
+      arrayBuffer = mapWriter(arrayDictionaryFile, arrayDictionaryWriter);
+      arrayDictionaryFile.toFile().deleteOnExit();
       arrayDictionary = FrontCodedIntArrayIndexed.read(arrayBuffer, ByteOrder.nativeOrder()).get();
       // reset position
       arrayBuffer.position(0);
+      arrayDictionaryFile.toFile().deleteOnExit();
     }
     final int index = arrayDictionary.indexOf(value);
     if (index < 0) {
@@ -213,15 +226,19 @@ public void close()
   {
     if (stringBufferMapper != null) {
       stringBufferMapper.close();
+      deleteTempFile(stringDictionaryFile);
     }
     if (longBuffer != null) {
       ByteBufferUtils.unmap(longBuffer);
+      deleteTempFile(longDictionaryFile);
     }
     if (doubleBuffer != null) {
       ByteBufferUtils.unmap(doubleBuffer);
+      deleteTempFile(doubleDictionaryFile);
     }
     if (arrayBuffer != null) {
       ByteBufferUtils.unmap(arrayBuffer);
+      deleteTempFile(arrayDictionaryFile);
     }
   }
 
@@ -250,6 +267,21 @@ private Path makeTempFile(String name)
     }
   }
 
+  private void deleteTempFile(Path path)
+  {
+    try {
+      final File file = path.toFile();
+      if (file.isDirectory()) {
+        FileUtils.deleteDirectory(file);
+      } else {
+        Files.delete(path);
+      }
+    }
+    catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   @SuppressFBWarnings("NP_NONNULL_PARAM_VIOLATION")
   private MappedByteBuffer mapWriter(Path path, DictionaryWriter<?> writer)
   {