Renamed 'Generic Column' -> 'Numeric Column'; Fixed a few resource leaks in processing; misc refinements

api/src/main/java/org/apache/druid/data/input/Row.java

@@ -24,6 +24,7 @@
 import org.apache.druid.guice.annotations.PublicApi;
 import org.joda.time.DateTime;
 
+import javax.annotation.Nullable;
 import java.util.List;
 
 /**
@@ -71,6 +72,7 @@ public interface Row extends Comparable<Row>
    *
    * @return the value of the provided column name
    */
+  @Nullable
   Object getRaw(String dimension);
 
   /**
@@ -79,5 +81,6 @@ public interface Row extends Comparable<Row>
    *  1. If the column is absent in the row, numeric zero is returned, rather than null.
    *  2. If the column has string value, an attempt is made to parse this value as a number.
    */
+  @Nullable
   Number getMetric(String metric);
 }

api/src/main/java/org/apache/druid/data/input/Rows.java

@@ -28,17 +28,16 @@
 import org.apache.druid.java.util.common.parsers.ParseException;
 
 import javax.annotation.Nullable;
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  */
 public class Rows
 {
-  public static final Long LONG_ZERO = 0L;
 
   /**
    * @param timeStamp rollup up timestamp to be used to create group key
@@ -55,10 +54,7 @@ public static List<Object> toGroupKey(long timeStamp, InputRow inputRow)
         dims.put(dim, dimValues);
       }
     }
-    return ImmutableList.of(
-        timeStamp,
-        dims
-    );
+    return ImmutableList.of(timeStamp, dims);
   }
 
   /**
@@ -70,14 +66,7 @@ public static List<String> objectToStrings(final Object inputValue)
       return Collections.emptyList();
     } else if (inputValue instanceof List) {
       // guava's toString function fails on null objects, so please do not use it
-      final List<Object> values = (List) inputValue;
-
-      final List<String> retVal = new ArrayList<>(values.size());
-      for (Object val : values) {
-        retVal.add(String.valueOf(val));
-      }
-
-      return retVal;
+      return ((List<?>) inputValue).stream().map(String::valueOf).collect(Collectors.toList());
     } else {
       return Collections.singletonList(String.valueOf(inputValue));
     }

api/src/main/java/org/apache/druid/data/input/impl/AbstractTextFilesFirehoseFactory.java

@@ -22,12 +22,12 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.io.LineIterator;
 import org.apache.druid.data.input.FiniteFirehoseFactory;
 import org.apache.druid.data.input.Firehose;
 import org.apache.druid.data.input.InputSplit;
 import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.io.LineIterator;
 
 import java.io.File;
 import java.io.IOException;

api/src/main/java/org/apache/druid/data/input/impl/FileIteratingFirehose.java

@@ -19,10 +19,10 @@
 
 package org.apache.druid.data.input.impl;
 
+import org.apache.commons.io.LineIterator;
 import org.apache.druid.data.input.Firehose;
 import org.apache.druid.data.input.InputRow;
 import org.apache.druid.utils.Runnables;
-import org.apache.commons.io.LineIterator;
 
 import javax.annotation.Nullable;
 import java.io.Closeable;
@@ -101,24 +101,9 @@ public Runnable commit()
   @Override
   public void close() throws IOException
   {
-    try {
-      if (lineIterator != null) {
-        lineIterator.close();
-      }
-    }
-    catch (Throwable t) {
-      try {
-        if (closer != null) {
-          closer.close();
-        }
-      }
-      catch (Exception e) {
-        t.addSuppressed(e);
-      }
-      throw t;
-    }
-    if (closer != null) {
-      closer.close();
+    try (Closeable ignore = closer;
+         Closeable ignore2 = lineIterator != null ? lineIterator::close : null) {
+      // close both via try-with-resources
     }
   }
 }

api/src/main/java/org/apache/druid/data/input/impl/prefetch/Fetcher.java

@@ -28,6 +28,8 @@
 import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Deque;
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
@@ -62,7 +64,7 @@ public abstract class Fetcher<T> implements Iterator<OpenedObject<T>>
   // This is updated when a file is successfully fetched, a fetched file is deleted, or a fetched file is
   // cached.
   private final AtomicLong fetchedBytes = new AtomicLong(0);
-  private Future<Void> fetchFuture;
+  private final Deque<Future<Void>> fetchFutures = new ArrayDeque<>();
   private PrefetchConfig prefetchConfig;
 
   // nextFetchIndex indicates which object should be downloaded when fetch is triggered.
@@ -103,12 +105,13 @@ public abstract class Fetcher<T> implements Iterator<OpenedObject<T>>
    */
   private void fetchIfNeeded(long remainingBytes)
   {
-    if ((fetchFuture == null || fetchFuture.isDone())
+    if ((fetchFutures.isEmpty() || fetchFutures.peekLast().isDone())
         && remainingBytes <= prefetchConfig.getPrefetchTriggerBytes()) {
-      fetchFuture = fetchExecutor.submit(() -> {
+      Future<Void> fetchFuture = fetchExecutor.submit(() -> {
         fetch();
         return null;
       });
+      fetchFutures.add(fetchFuture);
     }
   }
 
@@ -180,12 +183,17 @@ public OpenedObject<T> next()
   private void checkFetchException(boolean wait)
   {
     try {
-      if (wait) {
-        fetchFuture.get(prefetchConfig.getFetchTimeout(), TimeUnit.MILLISECONDS);
-        fetchFuture = null;
-      } else if (fetchFuture != null && fetchFuture.isDone()) {
-        fetchFuture.get();
-        fetchFuture = null;
+      for (Future<Void> fetchFuture; (fetchFuture = fetchFutures.poll()) != null; ) {
+        if (wait) {
+          fetchFuture.get(prefetchConfig.getFetchTimeout(), TimeUnit.MILLISECONDS);
+        } else {
+          if (fetchFuture.isDone()) {
+            fetchFuture.get();
+          } else {
+            fetchFutures.addFirst(fetchFuture);
+            break;
+          }
+        }
       }
     }
     catch (InterruptedException | ExecutionException e) {

api/src/main/java/org/apache/druid/data/input/impl/prefetch/FileFetcher.java

@@ -20,10 +20,9 @@
 package org.apache.druid.data.input.impl.prefetch;
 
 import com.google.common.base.Predicate;
-
+import org.apache.commons.io.IOUtils;
 import org.apache.druid.java.util.common.RetryUtils;
 import org.apache.druid.java.util.common.StringUtils;
-import org.apache.commons.io.IOUtils;
 
 import javax.annotation.Nullable;
 import java.io.Closeable;
@@ -33,7 +32,6 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.util.List;
-
 import java.util.concurrent.ExecutorService;
 
 /**

api/src/main/java/org/apache/druid/data/input/impl/prefetch/PrefetchSqlFirehoseFactory.java

@@ -21,18 +21,18 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
+import org.apache.commons.io.LineIterator;
 import org.apache.druid.data.input.Firehose;
 import org.apache.druid.data.input.FirehoseFactory;
 import org.apache.druid.data.input.impl.InputRowParser;
 import org.apache.druid.data.input.impl.SqlFirehose;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.logger.Logger;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.commons.io.LineIterator;
 
 import javax.annotation.Nullable;
 import java.io.File;

api/src/main/java/org/apache/druid/data/input/impl/prefetch/PrefetchableTextFilesFirehoseFactory.java

@@ -24,14 +24,14 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
+import org.apache.commons.io.LineIterator;
 import org.apache.druid.data.input.Firehose;
 import org.apache.druid.data.input.impl.AbstractTextFilesFirehoseFactory;
 import org.apache.druid.data.input.impl.FileIteratingFirehose;
 import org.apache.druid.data.input.impl.StringInputRowParser;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.commons.io.LineIterator;
 
 import javax.annotation.Nullable;
 import java.io.Closeable;
@@ -221,21 +221,24 @@ public LineIterator next()
             }
 
             final OpenedObject<T> openedObject = fetcher.next();
-            final InputStream stream;
             try {
-              stream = wrapObjectStream(
-                  openedObject.getObject(),
-                  openedObject.getObjectStream()
+              return new ResourceCloseableLineIterator(
+                  new InputStreamReader(
+                      wrapObjectStream(openedObject.getObject(), openedObject.getObjectStream()),
+                      StandardCharsets.UTF_8
+                  ),
+                  openedObject.getResourceCloser()
               );
             }
             catch (IOException e) {
+              try {
+                openedObject.getResourceCloser().close();
+              }
+              catch (Throwable t) {
+                e.addSuppressed(t);
+              }
               throw new RuntimeException(e);
             }
-
-            return new ResourceCloseableLineIterator(
-                new InputStreamReader(stream, StandardCharsets.UTF_8),
-                openedObject.getResourceCloser()
-            );
           }
         },
         firehoseParser,
@@ -288,9 +291,8 @@ static class ResourceCloseableLineIterator extends LineIterator
     @Override
     public void close()
     {
-      super.close();
-      try {
-        resourceCloser.close();
+      try (Closeable ignore = this.resourceCloser) {
+        super.close();
       }
       catch (IOException e) {
         throw new RuntimeException(e);

api/src/main/java/org/apache/druid/timeline/DataSegment.java

@@ -32,14 +32,14 @@
 import com.google.common.collect.Interner;
 import com.google.common.collect.Interners;
 import com.google.inject.Inject;
+import it.unimi.dsi.fastutil.objects.Object2ObjectArrayMap;
 import org.apache.druid.guice.annotations.PublicApi;
 import org.apache.druid.jackson.CommaListJoinDeserializer;
 import org.apache.druid.jackson.CommaListJoinSerializer;
 import org.apache.druid.java.util.common.granularity.Granularities;
 import org.apache.druid.query.SegmentDescriptor;
 import org.apache.druid.timeline.partition.NoneShardSpec;
 import org.apache.druid.timeline.partition.ShardSpec;
-import it.unimi.dsi.fastutil.objects.Object2ObjectArrayMap;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 

api/src/test/java/org/apache/druid/data/input/impl/FileIteratingFirehoseTest.java

@@ -22,8 +22,8 @@
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
-import org.apache.druid.data.input.InputRow;
 import org.apache.commons.io.LineIterator;
+import org.apache.druid.data.input.InputRow;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;

api/src/test/java/org/apache/druid/data/input/impl/JSONLowercaseParseSpecTest.java

@@ -20,8 +20,8 @@
 package org.apache.druid.data.input.impl;
 
 import com.google.common.collect.Lists;
-import org.apache.druid.java.util.common.parsers.Parser;
 import junit.framework.Assert;
+import org.apache.druid.java.util.common.parsers.Parser;
 import org.junit.Test;
 
 import java.util.Arrays;

api/src/test/java/org/apache/druid/data/input/impl/SqlFirehoseTest.java

@@ -26,9 +26,9 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
+import org.apache.commons.io.FileUtils;
 import org.apache.druid.data.input.InputRow;
 import org.apache.druid.data.input.impl.prefetch.JsonIterator;
-import org.apache.commons.io.FileUtils;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;