From 0701b1534d63ee73885f6a9be7439ebd94cb1cbb Mon Sep 17 00:00:00 2001
From: nishantmonu51 <nishant.monu51@gmail.com>
Date: Thu, 27 Feb 2014 14:39:25 +0530
Subject: [PATCH] review comments queryID changes, add wait time for realtime
 node queries

---
 .../main/java/io/druid/query/BaseQuery.java   | 15 +++++++-
 .../src/main/java/io/druid/query/Query.java   |  4 ++
 .../main/java/io/druid/query/QueryHelper.java | 37 -------------------
 .../groupby/GroupByQueryQueryToolChest.java   |  3 +-
 .../SegmentMetadataQueryQueryToolChest.java   |  3 +-
 .../search/SearchQueryQueryToolChest.java     | 26 +++++++------
 .../select/SelectQueryQueryToolChest.java     |  6 +--
 .../TimeBoundaryQueryQueryToolChest.java      |  3 +-
 .../TimeseriesQueryQueryToolChest.java        | 14 +++----
 .../query/topn/TopNQueryQueryToolChest.java   |  3 +-
 .../realtime/plumber/RealtimePlumber.java     |  2 +-
 .../java/io/druid/server/QueryIDProvider.java | 13 ++++++-
 .../java/io/druid/server/QueryResource.java   |  5 +--
 13 files changed, 58 insertions(+), 76 deletions(-)
 delete mode 100644 processing/src/main/java/io/druid/query/QueryHelper.java

diff --git a/processing/src/main/java/io/druid/query/BaseQuery.java b/processing/src/main/java/io/druid/query/BaseQuery.java
index b1e1a587123..7195e1fcd8e 100644
--- a/processing/src/main/java/io/druid/query/BaseQuery.java
+++ b/processing/src/main/java/io/druid/query/BaseQuery.java
@@ -21,6 +21,7 @@ package io.druid.query;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.metamx.common.guava.Sequence;
 import io.druid.query.spec.QuerySegmentSpec;
@@ -34,10 +35,10 @@ import java.util.Map;
  */
 public abstract class BaseQuery<T> implements Query<T>
 {
+  public static String QUERYID = "queryId";
   private final String dataSource;
   private final Map<String, String> context;
   private final QuerySegmentSpec querySegmentSpec;
-
   private volatile Duration duration;
 
   public BaseQuery(
@@ -130,4 +131,16 @@ public abstract class BaseQuery<T> implements Query<T>
 
     return overridden;
   }
+
+  @Override
+  public String getId()
+  {
+    return getContextValue(QUERYID);
+  }
+
+  @Override
+  public Query withId(String id)
+  {
+    return withOverriddenContext(ImmutableMap.of(QUERYID, id));
+  }
 }
diff --git a/processing/src/main/java/io/druid/query/Query.java b/processing/src/main/java/io/druid/query/Query.java
index 39da69c2b40..6823e1220ae 100644
--- a/processing/src/main/java/io/druid/query/Query.java
+++ b/processing/src/main/java/io/druid/query/Query.java
@@ -77,4 +77,8 @@ public interface Query<T>
   public Query<T> withOverriddenContext(Map<String, String> contextOverride);
 
   public Query<T> withQuerySegmentSpec(QuerySegmentSpec spec);
+
+  public Query<T> withId(String id);
+
+  public String getId();
 }
diff --git a/processing/src/main/java/io/druid/query/QueryHelper.java b/processing/src/main/java/io/druid/query/QueryHelper.java
deleted file mode 100644
index 4be72fd285f..00000000000
--- a/processing/src/main/java/io/druid/query/QueryHelper.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Druid - a distributed column store.
- * Copyright (C) 2012, 2013, 2014  Metamarkets Group Inc.
- *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License
- * as published by the Free Software Foundation; either version 2
- * of the License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
- */
-
-package io.druid.query;
-
-import com.google.common.collect.ImmutableMap;
-
-public class QueryHelper
-{
-  public static String QUERYID = "queryId";
-
-  public static String getQueryId(Query query)
-  {
-    return query.getContextValue(QUERYID);
-  }
-
-  public static Query setQueryId(Query query, String id)
-  {
-    return query.withOverriddenContext(ImmutableMap.of(QUERYID, id));
-  }
-}
diff --git a/processing/src/main/java/io/druid/query/groupby/GroupByQueryQueryToolChest.java b/processing/src/main/java/io/druid/query/groupby/GroupByQueryQueryToolChest.java
index 4eb727f726e..d2a338460e2 100644
--- a/processing/src/main/java/io/druid/query/groupby/GroupByQueryQueryToolChest.java
+++ b/processing/src/main/java/io/druid/query/groupby/GroupByQueryQueryToolChest.java
@@ -36,7 +36,6 @@ import io.druid.data.input.MapBasedRow;
 import io.druid.data.input.Row;
 import io.druid.query.IntervalChunkingQueryRunner;
 import io.druid.query.Query;
-import io.druid.query.QueryHelper;
 import io.druid.query.QueryRunner;
 import io.druid.query.QueryToolChest;
 import io.druid.query.aggregation.AggregatorFactory;
@@ -133,7 +132,7 @@ public class GroupByQueryQueryToolChest extends QueryToolChest<Row, GroupByQuery
         .setUser6(String.valueOf(query.hasFilters()))
         .setUser7(String.format("%,d aggs", query.getAggregatorSpecs().size()))
         .setUser9(Minutes.minutes(numMinutes).toString())
-        .setUser10(QueryHelper.getQueryId(query));
+        .setUser10(query.getId());
   }
 
   @Override
diff --git a/processing/src/main/java/io/druid/query/metadata/SegmentMetadataQueryQueryToolChest.java b/processing/src/main/java/io/druid/query/metadata/SegmentMetadataQueryQueryToolChest.java
index 41bfb21620a..00642baee3c 100644
--- a/processing/src/main/java/io/druid/query/metadata/SegmentMetadataQueryQueryToolChest.java
+++ b/processing/src/main/java/io/druid/query/metadata/SegmentMetadataQueryQueryToolChest.java
@@ -36,7 +36,6 @@ import io.druid.collections.OrderedMergeSequence;
 import io.druid.common.utils.JodaUtils;
 import io.druid.query.CacheStrategy;
 import io.druid.query.Query;
-import io.druid.query.QueryHelper;
 import io.druid.query.QueryRunner;
 import io.druid.query.QueryToolChest;
 import io.druid.query.ResultMergeQueryRunner;
@@ -153,7 +152,7 @@ public class SegmentMetadataQueryQueryToolChest extends QueryToolChest<SegmentAn
         .setUser5(Joiner.on(",").join(query.getIntervals()))
         .setUser6(String.valueOf(query.hasFilters()))
         .setUser9(Minutes.minutes(numMinutes).toString())
-        .setUser10(QueryHelper.getQueryId(query));
+        .setUser10(query.getId());
   }
 
   @Override
diff --git a/processing/src/main/java/io/druid/query/search/SearchQueryQueryToolChest.java b/processing/src/main/java/io/druid/query/search/SearchQueryQueryToolChest.java
index 1e36de6061b..6dd2f64e03b 100644
--- a/processing/src/main/java/io/druid/query/search/SearchQueryQueryToolChest.java
+++ b/processing/src/main/java/io/druid/query/search/SearchQueryQueryToolChest.java
@@ -40,7 +40,6 @@ import io.druid.collections.OrderedMergeSequence;
 import io.druid.query.CacheStrategy;
 import io.druid.query.IntervalChunkingQueryRunner;
 import io.druid.query.Query;
-import io.druid.query.QueryHelper;
 import io.druid.query.QueryRunner;
 import io.druid.query.QueryToolChest;
 import io.druid.query.Result;
@@ -66,11 +65,13 @@ import java.util.Set;
 public class SearchQueryQueryToolChest extends QueryToolChest<Result<SearchResultValue>, SearchQuery>
 {
   private static final byte SEARCH_QUERY = 0x2;
-
   private static final Joiner COMMA_JOIN = Joiner.on(",");
-  private static final TypeReference<Result<SearchResultValue>> TYPE_REFERENCE = new TypeReference<Result<SearchResultValue>>(){};
-
-  private static final TypeReference<Object> OBJECT_TYPE_REFERENCE = new TypeReference<Object>(){};
+  private static final TypeReference<Result<SearchResultValue>> TYPE_REFERENCE = new TypeReference<Result<SearchResultValue>>()
+  {
+  };
+  private static final TypeReference<Object> OBJECT_TYPE_REFERENCE = new TypeReference<Object>()
+  {
+  };
   private final SearchQueryConfig config;
 
   @Inject
@@ -125,7 +126,7 @@ public class SearchQueryQueryToolChest extends QueryToolChest<Result<SearchResul
         .setUser5(COMMA_JOIN.join(query.getIntervals()))
         .setUser6(String.valueOf(query.hasFilters()))
         .setUser9(Minutes.minutes(numMinutes).toString())
-        .setUser10(QueryHelper.getQueryId(query));
+        .setUser10(query.getId());
   }
 
   @Override
@@ -263,6 +264,11 @@ public class SearchQueryQueryToolChest extends QueryToolChest<Result<SearchResul
     );
   }
 
+  public Ordering<Result<SearchResultValue>> getOrdering()
+  {
+    return Ordering.natural();
+  }
+
   private static class SearchThresholdAdjustingQueryRunner implements QueryRunner<Result<SearchResultValue>>
   {
     private final QueryRunner<Result<SearchResultValue>> runner;
@@ -271,7 +277,8 @@ public class SearchQueryQueryToolChest extends QueryToolChest<Result<SearchResul
     public SearchThresholdAdjustingQueryRunner(
         QueryRunner<Result<SearchResultValue>> runner,
         SearchQueryConfig config
-    ) {
+    )
+    {
       this.runner = runner;
       this.config = config;
     }
@@ -343,9 +350,4 @@ public class SearchQueryQueryToolChest extends QueryToolChest<Result<SearchResul
       );
     }
   }
-
-  public Ordering<Result<SearchResultValue>> getOrdering()
-  {
-    return Ordering.natural();
-  }
 }
diff --git a/processing/src/main/java/io/druid/query/select/SelectQueryQueryToolChest.java b/processing/src/main/java/io/druid/query/select/SelectQueryQueryToolChest.java
index 03ba9adeec4..8a728749b22 100644
--- a/processing/src/main/java/io/druid/query/select/SelectQueryQueryToolChest.java
+++ b/processing/src/main/java/io/druid/query/select/SelectQueryQueryToolChest.java
@@ -37,7 +37,6 @@ import io.druid.query.CacheStrategy;
 import io.druid.query.IntervalChunkingQueryRunner;
 import io.druid.query.Query;
 import io.druid.query.QueryConfig;
-import io.druid.query.QueryHelper;
 import io.druid.query.QueryRunner;
 import io.druid.query.QueryToolChest;
 import io.druid.query.Result;
@@ -61,18 +60,15 @@ import java.util.Set;
 public class SelectQueryQueryToolChest extends QueryToolChest<Result<SelectResultValue>, SelectQuery>
 {
   private static final byte SELECT_QUERY = 0x13;
-
   private static final Joiner COMMA_JOIN = Joiner.on(",");
   private static final TypeReference<Object> OBJECT_TYPE_REFERENCE =
       new TypeReference<Object>()
       {
       };
-
   private static final TypeReference<Result<SelectResultValue>> TYPE_REFERENCE =
       new TypeReference<Result<SelectResultValue>>()
       {
       };
-
   private final QueryConfig config;
   private final ObjectMapper jsonMapper;
 
@@ -132,7 +128,7 @@ public class SelectQueryQueryToolChest extends QueryToolChest<Result<SelectResul
         .setUser5(COMMA_JOIN.join(query.getIntervals()))
         .setUser6(String.valueOf(query.hasFilters()))
         .setUser9(Minutes.minutes(numMinutes).toString())
-        .setUser10(QueryHelper.getQueryId(query));
+        .setUser10(query.getId());
   }
 
   @Override
diff --git a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryQueryToolChest.java b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryQueryToolChest.java
index 0dc126db029..7ccba774799 100644
--- a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryQueryToolChest.java
+++ b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryQueryToolChest.java
@@ -34,7 +34,6 @@ import io.druid.collections.OrderedMergeSequence;
 import io.druid.query.BySegmentSkippingQueryRunner;
 import io.druid.query.CacheStrategy;
 import io.druid.query.Query;
-import io.druid.query.QueryHelper;
 import io.druid.query.QueryRunner;
 import io.druid.query.QueryToolChest;
 import io.druid.query.Result;
@@ -121,7 +120,7 @@ public class TimeBoundaryQueryQueryToolChest
         .setUser2(query.getDataSource())
         .setUser4(query.getType())
         .setUser6("false")
-        .setUser10(QueryHelper.getQueryId(query));
+        .setUser10(query.getId());
   }
 
   @Override
diff --git a/processing/src/main/java/io/druid/query/timeseries/TimeseriesQueryQueryToolChest.java b/processing/src/main/java/io/druid/query/timeseries/TimeseriesQueryQueryToolChest.java
index 2ac322fe8d9..d5de1bda272 100644
--- a/processing/src/main/java/io/druid/query/timeseries/TimeseriesQueryQueryToolChest.java
+++ b/processing/src/main/java/io/druid/query/timeseries/TimeseriesQueryQueryToolChest.java
@@ -37,7 +37,6 @@ import io.druid.query.IntervalChunkingQueryRunner;
 import io.druid.query.Query;
 import io.druid.query.QueryCacheHelper;
 import io.druid.query.QueryConfig;
-import io.druid.query.QueryHelper;
 import io.druid.query.QueryRunner;
 import io.druid.query.QueryToolChest;
 import io.druid.query.Result;
@@ -62,14 +61,15 @@ import java.util.Map;
 public class TimeseriesQueryQueryToolChest extends QueryToolChest<Result<TimeseriesResultValue>, TimeseriesQuery>
 {
   private static final byte TIMESERIES_QUERY = 0x0;
-
   private static final Joiner COMMA_JOIN = Joiner.on(",");
   private static final TypeReference<Object> OBJECT_TYPE_REFERENCE =
-      new TypeReference<Object>(){};
-
+      new TypeReference<Object>()
+      {
+      };
   private static final TypeReference<Result<TimeseriesResultValue>> TYPE_REFERENCE =
-      new TypeReference<Result<TimeseriesResultValue>>() {};
-
+      new TypeReference<Result<TimeseriesResultValue>>()
+      {
+      };
   private final QueryConfig config;
 
   @Inject
@@ -129,7 +129,7 @@ public class TimeseriesQueryQueryToolChest extends QueryToolChest<Result<Timeser
         .setUser6(String.valueOf(query.hasFilters()))
         .setUser7(String.format("%,d aggs", query.getAggregatorSpecs().size()))
         .setUser9(Minutes.minutes(numMinutes).toString())
-        .setUser10(QueryHelper.getQueryId(query));
+        .setUser10(query.getId());
   }
 
   @Override
diff --git a/processing/src/main/java/io/druid/query/topn/TopNQueryQueryToolChest.java b/processing/src/main/java/io/druid/query/topn/TopNQueryQueryToolChest.java
index 27d83a50205..dc4dd1b4855 100644
--- a/processing/src/main/java/io/druid/query/topn/TopNQueryQueryToolChest.java
+++ b/processing/src/main/java/io/druid/query/topn/TopNQueryQueryToolChest.java
@@ -40,7 +40,6 @@ import io.druid.query.CacheStrategy;
 import io.druid.query.IntervalChunkingQueryRunner;
 import io.druid.query.Query;
 import io.druid.query.QueryCacheHelper;
-import io.druid.query.QueryHelper;
 import io.druid.query.QueryRunner;
 import io.druid.query.QueryToolChest;
 import io.druid.query.Result;
@@ -135,7 +134,7 @@ public class TopNQueryQueryToolChest extends QueryToolChest<Result<TopNResultVal
         .setUser6(String.valueOf(query.hasFilters()))
         .setUser7(String.format("%,d aggs", query.getAggregatorSpecs().size()))
         .setUser9(Minutes.minutes(numMinutes).toString())
-        .setUser10(QueryHelper.getQueryId(query));
+        .setUser10(query.getId());
   }
 
   @Override
diff --git a/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java b/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java
index 62b3d3d5f6d..9b3e5652662 100644
--- a/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java
+++ b/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java
@@ -257,7 +257,7 @@ public class RealtimePlumber implements Plumber
                                         }
                                     )
                                 )
-                            ),
+                            ).withWaitMeasuredFromNow(),
                             new SpecificSegmentSpec(
                                 new SegmentDescriptor(
                                     holder.getInterval(),
diff --git a/server/src/main/java/io/druid/server/QueryIDProvider.java b/server/src/main/java/io/druid/server/QueryIDProvider.java
index 03504373a31..2951ec02726 100644
--- a/server/src/main/java/io/druid/server/QueryIDProvider.java
+++ b/server/src/main/java/io/druid/server/QueryIDProvider.java
@@ -21,6 +21,8 @@ package io.druid.server;
 
 import com.google.inject.Inject;
 import io.druid.guice.annotations.Self;
+import io.druid.query.Query;
+import org.joda.time.DateTime;
 
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -35,8 +37,15 @@ public class QueryIdProvider
     host = node.getHost();
   }
 
-  public String next()
+  public String next(Query query)
   {
-    return String.format("%s_%s", host, id.incrementAndGet());
+    return String.format(
+        "%s_%s_%s_%s_%s",
+        query.getDataSource(),
+        query.getDuration(),
+        host,
+        new DateTime(),
+        id.incrementAndGet()
+    );
   }
 }
diff --git a/server/src/main/java/io/druid/server/QueryResource.java b/server/src/main/java/io/druid/server/QueryResource.java
index 353cf9a15dd..a7741dc0adb 100644
--- a/server/src/main/java/io/druid/server/QueryResource.java
+++ b/server/src/main/java/io/druid/server/QueryResource.java
@@ -35,7 +35,6 @@ import com.metamx.emitter.service.ServiceMetricEvent;
 import io.druid.guice.annotations.Json;
 import io.druid.guice.annotations.Smile;
 import io.druid.query.Query;
-import io.druid.query.QueryHelper;
 import io.druid.query.QuerySegmentWalker;
 import io.druid.server.log.RequestLogger;
 import org.joda.time.DateTime;
@@ -106,9 +105,9 @@ public class QueryResource
     try {
       requestQuery = ByteStreams.toByteArray(req.getInputStream());
       query = objectMapper.readValue(requestQuery, Query.class);
-      queryID = QueryHelper.getQueryId(query);
+      queryID = query.getId();
       if (queryID == null) {
-        query = QueryHelper.setQueryId(query, idProvider.next());
+        query = query.withId(idProvider.next(query));
       }
 
       requestLogger.log(