Better handle null adapters and NPEs in the CQE

pom.xml

@@ -74,7 +74,7 @@
             <dependency>
                 <groupId>com.metamx</groupId>
                 <artifactId>emitter</artifactId>
-                <version>0.2.10</version>
+                <version>0.2.11</version>
             </dependency>
             <dependency>
                 <groupId>com.metamx</groupId>

processing/src/main/java/io/druid/query/ChainedExecutionQueryRunner.java

@@ -35,8 +35,10 @@ import com.metamx.common.logger.Logger;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 /**
@@ -110,7 +112,11 @@ public class ChainedExecutionQueryRunner<T> implements QueryRunner<T>
                                   if (input == null) {
                                     throw new ISE("Input is null?! How is this possible?!");
                                   }
-                                  return Sequences.toList(input.run(query), Lists.<T>newArrayList());
+                                  Sequence<T> result = input.run(query);
+                                  if (result == null) {
+                                    throw new ISE("Got a null result! Segments are missing!");
+                                  }
+                                  return Sequences.toList(result, Lists.<T>newArrayList());
                                 }
                                 catch (Exception e) {
                                   log.error(e, "Exception with one of the sequences!");
@@ -156,4 +162,20 @@ public class ChainedExecutionQueryRunner<T> implements QueryRunner<T>
         }
     );
   }
+
+  public static void main(String[] args) throws Exception
+  {
+    ExecutorService foo = Executors.newFixedThreadPool(1);
+    Future test = foo.submit(
+        new Callable<List>()
+        {
+          @Override
+          public List call() throws Exception
+          {
+            throw new ISE("");
+          }
+        }
+    );
+    System.out.println(Lists.newArrayList(test));
+  }
 }

processing/src/main/java/io/druid/query/groupby/GroupByQueryEngine.java

@@ -70,7 +70,7 @@ public class GroupByQueryEngine
   private final StupidPool<ByteBuffer> intermediateResultsBufferPool;
 
   @Inject
-  public GroupByQueryEngine (
+  public GroupByQueryEngine(
       Supplier<GroupByQueryConfig> config,
       @Global StupidPool<ByteBuffer> intermediateResultsBufferPool
   )
@@ -81,6 +81,12 @@ public class GroupByQueryEngine
 
   public Sequence<Row> process(final GroupByQuery query, StorageAdapter storageAdapter)
   {
+    if (storageAdapter == null) {
+      throw new ISE(
+          "Null storage adapter found. Probably trying to issue a query against a segment being memory unmapped."
+      );
+    }
+
     final List<Interval> intervals = query.getQuerySegmentSpec().getIntervals();
     if (intervals.size() != 1) {
       throw new IAE("Should only have one interval, got[%s]", intervals);
@@ -187,8 +193,7 @@ public class GroupByQueryEngine
           ByteBuffer newKey = key.duplicate();
           newKey.putInt(dimSelector.getValueCardinality());
           unaggregatedBuffers = updateValues(newKey, dims.subList(1, dims.size()));
-        }
+        } else {
-        else {
           for (Integer dimValue : row) {
             ByteBuffer newKey = key.duplicate();
             newKey.putInt(dimValue);
@@ -202,8 +207,7 @@ public class GroupByQueryEngine
           retVal.addAll(unaggregatedBuffers);
         }
         return retVal;
-      }
+      } else {
-      else {
         key.clear();
         Integer position = positions.get(key);
         int[] increments = positionMaintainer.getIncrements();
@@ -267,8 +271,7 @@ public class GroupByQueryEngine
     {
       if (nextVal > max) {
         return null;
-      }
+      } else {
-      else {
         int retVal = (int) nextVal;
         nextVal += increment;
         return retVal;
@@ -402,7 +405,7 @@ public class GroupByQueryEngine
                     final DimExtractionFn fn = dimensionSpecs.get(i).getDimExtractionFn();
                     final int dimVal = keyBuffer.getInt();
                     if (dimSelector.getValueCardinality() != dimVal) {
-                      if(fn != null) {
+                      if (fn != null) {
                         theEvent.put(dimNames.get(i), fn.apply(dimSelector.lookupName(dimVal)));
                       } else {
                         theEvent.put(dimNames.get(i), dimSelector.lookupName(dimVal));
@@ -434,9 +437,10 @@ public class GroupByQueryEngine
       throw new UnsupportedOperationException();
     }
 
-    public void close() {
+    public void close()
+    {
       // cleanup
-      for(BufferAggregator agg : aggregators) {
+      for (BufferAggregator agg : aggregators) {
         agg.close();
       }
     }

processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java

@@ -32,7 +32,6 @@ import java.util.Map;
 
 public class SegmentMetadataQuery extends BaseQuery<SegmentAnalysis>
 {
-
   private final ColumnIncluderator toInclude;
   private final boolean merge;
 

processing/src/main/java/io/druid/query/search/SearchQueryRunner.java

@@ -55,7 +55,7 @@ import java.util.Map;
 import java.util.TreeSet;
 
 /**
-*/
+ */
 public class SearchQueryRunner implements QueryRunner<Result<SearchResultValue>>
 {
   private static final EmittingLogger log = new EmittingLogger(SearchQueryRunner.class);
@@ -99,12 +99,10 @@ public class SearchQueryRunner implements QueryRunner<Result<SearchResultValue>>
           ConciseSet set = new ConciseSet();
           set.add(0);
           baseFilter = ImmutableConciseSet.newImmutableFromMutable(set);
-        }
+        } else {
-        else {
           baseFilter = ImmutableConciseSet.complement(new ImmutableConciseSet(), index.getNumRows());
         }
-      }
+      } else {
-      else {
         baseFilter = filter.goConcise(new ColumnSelectorBitmapIndexSelector(index));
       }
 
@@ -133,7 +131,16 @@ public class SearchQueryRunner implements QueryRunner<Result<SearchResultValue>>
     }
 
     final StorageAdapter adapter = segment.asStorageAdapter();
-    if (adapter != null) {
+
+    if (adapter == null) {
+      log.makeAlert("WTF!? Unable to process search query on segment.")
+         .addData("segment", segment.getIdentifier())
+         .addData("query", query).emit();
+      throw new ISE(
+          "Null storage adapter found. Probably trying to issue a query against a segment being memory unmapped."
+      );
+    }
+
     Iterable<String> dimsToSearch;
     if (dimensions == null || dimensions.isEmpty()) {
       dimsToSearch = adapter.getAvailableDimensions();
@@ -172,12 +179,6 @@ public class SearchQueryRunner implements QueryRunner<Result<SearchResultValue>>
     return makeReturnResult(limit, retVal);
   }
 
-    log.makeAlert("WTF!? Unable to process search query on segment.")
-       .addData("segment", segment.getIdentifier())
-       .addData("query", query);
-    return Sequences.empty();
-  }
-
   private Sequence<Result<SearchResultValue>> makeReturnResult(int limit, TreeSet<SearchHit> retVal)
   {
     return Sequences.simple(

processing/src/main/java/io/druid/query/select/SelectQueryEngine.java

@@ -22,6 +22,7 @@ package io.druid.query.select;
 import com.google.common.base.Function;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.metamx.common.ISE;
 import com.metamx.common.guava.BaseSequence;
 import com.metamx.common.guava.Sequence;
 import io.druid.query.QueryRunnerHelper;
@@ -54,6 +55,12 @@ public class SelectQueryEngine
           {
             final StorageAdapter adapter = segment.asStorageAdapter();
 
+            if (adapter == null) {
+              throw new ISE(
+                  "Null storage adapter found. Probably trying to issue a query against a segment being memory unmapped."
+              );
+            }
+
             final Iterable<String> dims;
             if (query.getDimensions() == null || query.getDimensions().isEmpty()) {
               dims = adapter.getAvailableDimensions();

processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryRunnerFactory.java

@@ -87,6 +87,12 @@ public class TimeBoundaryQueryRunnerFactory
             @Override
             public Iterator<Result<TimeBoundaryResultValue>> make()
             {
+              if (adapter == null) {
+                throw new ISE(
+                    "Null storage adapter found. Probably trying to issue a query against a segment being memory unmapped."
+                );
+              }
+
               return legacyQuery.buildResult(
                   adapter.getInterval().getStart(),
                   adapter.getMinTime(),

processing/src/main/java/io/druid/query/timeseries/TimeseriesQueryEngine.java

@@ -20,6 +20,7 @@
 package io.druid.query.timeseries;
 
 import com.google.common.base.Function;
+import com.metamx.common.ISE;
 import com.metamx.common.guava.BaseSequence;
 import com.metamx.common.guava.Sequence;
 import io.druid.query.QueryRunnerHelper;
@@ -40,6 +41,12 @@ public class TimeseriesQueryEngine
 {
   public Sequence<Result<TimeseriesResultValue>> process(final TimeseriesQuery query, final StorageAdapter adapter)
   {
+    if (adapter == null) {
+      throw new ISE(
+          "Null storage adapter found. Probably trying to issue a query against a segment being memory unmapped."
+      );
+    }
+
     return new BaseSequence<Result<TimeseriesResultValue>, Iterator<Result<TimeseriesResultValue>>>(
         new BaseSequence.IteratorMaker<Result<TimeseriesResultValue>, Iterator<Result<TimeseriesResultValue>>>()
         {

processing/src/main/java/io/druid/query/topn/TopNQueryEngine.java

@@ -21,7 +21,7 @@ package io.druid.query.topn;
 
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
+import com.metamx.common.ISE;
 import com.metamx.common.guava.FunctionalIterable;
 import com.metamx.common.logger.Logger;
 import io.druid.collections.StupidPool;
@@ -53,6 +53,12 @@ public class TopNQueryEngine
 
   public Iterable<Result<TopNResultValue>> query(final TopNQuery query, final StorageAdapter adapter)
   {
+    if (adapter == null) {
+      throw new ISE(
+          "Null storage adapter found. Probably trying to issue a query against a segment being memory unmapped."
+      );
+    }
+
     final List<Interval> queryIntervals = query.getQuerySegmentSpec().getIntervals();
     final Filter filter = Filters.convertDimensionFilters(query.getDimensionsFilter());
     final QueryGranularity granularity = query.getGranularity();
@@ -62,10 +68,6 @@ public class TopNQueryEngine
         queryIntervals.size() == 1, "Can only handle a single interval, got[%s]", queryIntervals
     );
 
-    if (mapFn == null) {
-      return Lists.newArrayList();
-    }
-
     return FunctionalIterable
         .create(adapter.makeCursors(filter, queryIntervals.get(0), granularity))
         .transform(
@@ -84,13 +86,6 @@ public class TopNQueryEngine
 
   private Function<Cursor, Result<TopNResultValue>> getMapFn(TopNQuery query, final StorageAdapter adapter)
   {
-    if (adapter == null) {
-      log.warn(
-          "Null storage adapter found. Probably trying to issue a query against a segment being memory unmapped. Returning empty results."
-      );
-      return null;
-    }
-
     final Capabilities capabilities = adapter.getCapabilities();
     final int cardinality = adapter.getDimensionCardinality(query.getDimensionSpec().getDimension());
     int numBytesPerRecord = 0;