[Backport] Druid quickstart: Update task memory

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

@@ -53,6 +53,12 @@
 
 public class FileUtils
 {
+  public enum LinkOrCopyResult
+  {
+    LINK,
+    COPY
+  }
+
   /**
    * Useful for retry functionality that doesn't want to stop Throwables, but does want to retry on Exceptions
    */
@@ -461,6 +467,26 @@ public static void deleteDirectory(final File directory) throws IOException
     org.apache.commons.io.FileUtils.deleteDirectory(directory);
   }
 
+  /**
+   * Hard-link "src" as "dest", if possible. If not possible -- perhaps they are on separate filesystems -- then
+   * copy "src" to "dest".
+   *
+   * @return whether a link or copy was made. Can be safely ignored if you don't care.
+   *
+   * @throws IOException if something went wrong
+   */
+  public static LinkOrCopyResult linkOrCopy(final File src, final File dest) throws IOException
+  {
+    try {
+      Files.createLink(dest.toPath(), src.toPath());
+      return LinkOrCopyResult.LINK;
+    }
+    catch (IOException e) {
+      Files.copy(src.toPath(), dest.toPath(), StandardCopyOption.REPLACE_EXISTING);
+      return LinkOrCopyResult.COPY;
+    }
+  }
+
   public interface OutputStreamConsumer<T>
   {
     T apply(OutputStream outputStream) throws IOException;

core/src/main/java/org/apache/druid/java/util/common/Intervals.java

@@ -20,10 +20,13 @@
 package org.apache.druid.java.util.common;
 
 import com.google.common.collect.ImmutableList;
+import org.apache.druid.java.util.common.guava.Comparators;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.joda.time.chrono.ISOChronology;
 
+import javax.annotation.Nullable;
+
 public final class Intervals
 {
   public static final Interval ETERNITY = utc(JodaUtils.MIN_INSTANT, JodaUtils.MAX_INSTANT);
@@ -68,6 +71,32 @@ public static boolean isEternity(final Interval interval)
     return ETERNITY.equals(interval);
   }
 
+  /**
+   * Finds an interval from the given set of sortedIntervals which overlaps with
+   * the searchInterval. If multiple candidate intervals overlap with the
+   * searchInterval, the "smallest" interval based on the
+   * {@link Comparators#intervalsByStartThenEnd()} is returned.
+   *
+   * @param searchInterval  Interval which should overlap with the result
+   * @param sortedIntervals Candidate overlapping intervals, sorted in ascending
+   *                        order, using {@link Comparators#intervalsByStartThenEnd()}.
+   * @return The first overlapping interval, if one exists, otherwise null.
+   */
+  @Nullable
+  public static Interval findOverlappingInterval(Interval searchInterval, Interval[] sortedIntervals)
+  {
+    for (Interval interval : sortedIntervals) {
+      if (interval.overlaps(searchInterval)) {
+        return interval;
+      } else if (interval.getStart().isAfter(searchInterval.getEnd())) {
+        // Intervals after this cannot have an overlap
+        return null;
+      }
+    }
+
+    return null;
+  }
+
   private Intervals()
   {
   }

core/src/main/java/org/apache/druid/java/util/common/parsers/FlattenerJsonProvider.java

@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.java.util.common.parsers;
+
+import com.jayway.jsonpath.InvalidJsonException;
+import com.jayway.jsonpath.spi.json.JsonProvider;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+public abstract class FlattenerJsonProvider implements JsonProvider
+{
+  @Override
+  public Object createArray()
+  {
+    return new ArrayList<>();
+  }
+
+  @Override
+  public Object createMap()
+  {
+    return new LinkedHashMap<>();
+  }
+
+  @Override
+  public int length(final Object o)
+  {
+    if (o instanceof List) {
+      return ((List) o).size();
+    } else {
+      return 0;
+    }
+  }
+
+  @Override
+  public Iterable<?> toIterable(final Object o)
+  {
+    if (o instanceof List) {
+      return (List) o;
+    }
+    throw new UnsupportedOperationException(o.getClass().getName());
+  }
+
+  @Override
+  public Object getArrayIndex(final Object o, final int i)
+  {
+    if (o instanceof List) {
+      return ((List) o).get(i);
+    }
+    throw new UnsupportedOperationException(o.getClass().getName());
+  }
+
+  @Override
+  public void setArrayIndex(final Object o, final int i, final Object o1)
+  {
+    if (o instanceof List) {
+      final List list = (List) o;
+      if (list.size() == i) {
+        list.add(o1);
+      } else {
+        list.set(i, o1);
+      }
+    } else {
+      throw new UnsupportedOperationException(o.getClass().getName());
+    }
+  }
+
+  @Override
+  public void setProperty(final Object o, final Object o1, final Object o2)
+  {
+    if (o instanceof Map) {
+      ((Map) o).put(o1, o2);
+    } else {
+      throw new UnsupportedOperationException(o.getClass().getName());
+    }
+  }
+
+  @Override
+  public void removeProperty(final Object o, final Object o1)
+  {
+    if (o instanceof Map) {
+      ((Map) o).remove(o1);
+    } else {
+      throw new UnsupportedOperationException(o.getClass().getName());
+    }
+  }
+
+  @Override
+  @Deprecated
+  public Object getArrayIndex(final Object o, final int i, final boolean b)
+  {
+    throw new UnsupportedOperationException("Deprecated");
+  }
+
+  @Override
+  public Object parse(final String s) throws InvalidJsonException
+  {
+    throw new UnsupportedOperationException("Unused");
+  }
+
+  @Override
+  public Object parse(final InputStream inputStream, final String s) throws InvalidJsonException
+  {
+    throw new UnsupportedOperationException("Unused");
+  }
+
+  @Override
+  public String toJson(final Object o)
+  {
+    throw new UnsupportedOperationException("Unused");
+  }
+
+  @Override
+  public Object unwrap(final Object o)
+  {
+    return o;
+  }
+}

core/src/main/java/org/apache/druid/java/util/common/parsers/JSONFlattenerMaker.java

@@ -54,9 +54,9 @@ public class JSONFlattenerMaker implements ObjectFlatteners.FlattenerMaker<JsonN
                    .options(EnumSet.of(Option.SUPPRESS_EXCEPTIONS))
                    .build();
 
+  private final CharsetEncoder enc = StandardCharsets.UTF_8.newEncoder();
   private final boolean keepNullValues;
 
-  private final CharsetEncoder enc = StandardCharsets.UTF_8.newEncoder();
 
   public JSONFlattenerMaker(boolean keepNullValues)
   {
@@ -66,7 +66,7 @@ public JSONFlattenerMaker(boolean keepNullValues)
   @Override
   public Iterable<String> discoverRootFields(final JsonNode obj)
   {
-    return FluentIterable.from(() -> obj.fields())
+    return FluentIterable.from(obj::fields)
                          .filter(
                              entry -> {
                                final JsonNode val = entry.getValue();
@@ -137,13 +137,13 @@ public JsonProvider getJsonProvider()
   public Object finalizeConversionForMap(Object o)
   {
     if (o instanceof JsonNode) {
-      return convertJsonNode((JsonNode) o);
+      return convertJsonNode((JsonNode) o, enc);
     }
     return o;
   }
 
   @Nullable
-  private Object convertJsonNode(JsonNode val)
+  public static Object convertJsonNode(JsonNode val, CharsetEncoder enc)
   {
     if (val == null || val.isNull()) {
       return null;
@@ -158,7 +158,7 @@ private Object convertJsonNode(JsonNode val)
     }
 
     if (val.isTextual()) {
-      return charsetFix(val.asText());
+      return charsetFix(val.asText(), enc);
     }
 
     if (val.isBoolean()) {
@@ -175,7 +175,7 @@ private Object convertJsonNode(JsonNode val)
       List<Object> newList = new ArrayList<>();
       for (JsonNode entry : val) {
         if (!entry.isNull()) {
-          newList.add(finalizeConversionForMap(entry));
+          newList.add(convertJsonNode(entry, enc));
         }
       }
       return newList;
@@ -185,7 +185,7 @@ private Object convertJsonNode(JsonNode val)
       Map<String, Object> newMap = new LinkedHashMap<>();
       for (Iterator<Map.Entry<String, JsonNode>> it = val.fields(); it.hasNext(); ) {
         Map.Entry<String, JsonNode> entry = it.next();
-        newMap.put(entry.getKey(), finalizeConversionForMap(entry.getValue()));
+        newMap.put(entry.getKey(), convertJsonNode(entry.getValue(), enc));
       }
       return newMap;
     }
@@ -197,7 +197,7 @@ private Object convertJsonNode(JsonNode val)
   }
 
   @Nullable
-  private String charsetFix(String s)
+  private static String charsetFix(String s, CharsetEncoder enc)
   {
     if (s != null && !enc.canEncode(s)) {
       // Some whacky characters are in this string (e.g. \uD900). These are problematic because they are decodeable
@@ -209,7 +209,7 @@ private String charsetFix(String s)
     }
   }
 
-  private boolean isFlatList(JsonNode list)
+  private static boolean isFlatList(JsonNode list)
   {
     for (JsonNode obj : list) {
       if (obj.isObject() || obj.isArray()) {

core/src/main/java/org/apache/druid/java/util/common/parsers/ObjectFlatteners.java

@@ -29,6 +29,7 @@
 import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
@@ -250,15 +251,23 @@ default Function<T, Object> makeJsonTreeExtractor(List<String> nodes)
      */
     default Map<String, Object> toMap(T obj)
     {
-      return (Map<String, Object>) toPlainJavaType(obj);
+      final Object mapOrNull = toPlainJavaType(obj);
+      if (mapOrNull == null) {
+        return Collections.emptyMap();
+      }
+      return (Map<String, Object>) mapOrNull;
     }
 
     /**
      * Recursively traverse "json" object using a {@link JsonProvider}, converting to Java {@link Map} and {@link List},
      * potentially transforming via {@link #finalizeConversionForMap} as we go
      */
+    @Nullable
     default Object toPlainJavaType(Object o)
     {
+      if (o == null) {
+        return null;
+      }
       final JsonProvider jsonProvider = getJsonProvider();
       if (jsonProvider.isMap(o)) {
         Map<String, Object> actualMap = new HashMap<>();
@@ -287,7 +296,7 @@ default Object toPlainJavaType(Object o)
         return finalizeConversionForMap(actualList);
       }
       // unknown, just pass it through
-      return o;
+      return finalizeConversionForMap(o);
     }
 
     /**

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

@@ -244,4 +244,45 @@ public void testCopyLarge() throws IOException
     Assert.assertEquals(data.length(), result);
     Assert.assertEquals(data, StringUtils.fromUtf8(Files.readAllBytes(dstFile.toPath())));
   }
+
+  @Test
+  public void testLinkOrCopy1() throws IOException
+  {
+    // Will be a LINK.
+
+    final File fromFile = temporaryFolder.newFile();
+    final File toDir = temporaryFolder.newFolder();
+    final File toFile = new File(toDir, "toFile");
+
+    Files.write(fromFile.toPath(), StringUtils.toUtf8("foo"));
+    final FileUtils.LinkOrCopyResult linkOrCopyResult = FileUtils.linkOrCopy(fromFile, toFile);
+
+    // Verify the new link.
+    Assert.assertEquals(FileUtils.LinkOrCopyResult.LINK, linkOrCopyResult);
+    Assert.assertEquals("foo", StringUtils.fromUtf8(Files.readAllBytes(toFile.toPath())));
+
+    // Verify they are actually the same file.
+    Files.write(fromFile.toPath(), StringUtils.toUtf8("bar"));
+    Assert.assertEquals("bar", StringUtils.fromUtf8(Files.readAllBytes(toFile.toPath())));
+  }
+
+  @Test
+  public void testLinkOrCopy2() throws IOException
+  {
+    // Will be a COPY, because the destination file already exists and therefore Files.createLink fails.
+
+    final File fromFile = temporaryFolder.newFile();
+    final File toFile = temporaryFolder.newFile();
+
+    Files.write(fromFile.toPath(), StringUtils.toUtf8("foo"));
+    final FileUtils.LinkOrCopyResult linkOrCopyResult = FileUtils.linkOrCopy(fromFile, toFile);
+
+    // Verify the new link.
+    Assert.assertEquals(FileUtils.LinkOrCopyResult.COPY, linkOrCopyResult);
+    Assert.assertEquals("foo", StringUtils.fromUtf8(Files.readAllBytes(toFile.toPath())));
+
+    // Verify they are not the same file.
+    Files.write(fromFile.toPath(), StringUtils.toUtf8("bar"));
+    Assert.assertEquals("foo", StringUtils.fromUtf8(Files.readAllBytes(toFile.toPath())));
+  }
 }