Help relieve some memory pressure on brokers & coordinator

api/src/main/java/io/druid/timeline/DataSegment.java

@@ -23,14 +23,15 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Interner;
 import com.google.common.collect.Interners;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Ordering;
 import com.metamx.common.Granularity;
 import io.druid.jackson.CommaListJoinDeserializer;
 import io.druid.jackson.CommaListJoinSerializer;
@@ -40,6 +41,7 @@
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
@@ -50,15 +52,8 @@ public class DataSegment implements Comparable<DataSegment>
 {
   public static String delimiter = "_";
   private final Integer binaryVersion;
-  private static final Interner<String> interner = Interners.newWeakInterner();
-  private static final Function<String, String> internFun = new Function<String, String>()
-  {
-    @Override
-    public String apply(String input)
-    {
-      return interner.intern(input);
-    }
-  };
+  private static final Interner<String> STRING_INTERNER = Interners.newWeakInterner();
+  private static final Interner<ImmutableList<String>> LIST_INTERNER = Interners.newWeakInterner();
 
   public static String makeDataSegmentIdentifier(
       String dataSource,
@@ -116,17 +111,19 @@ public boolean apply(String input)
     };
 
     // dataSource, dimensions & metrics are stored as canonical string values to decrease memory required for storing large numbers of segments.
-    this.dataSource = interner.intern(dataSource);
+    this.dataSource = STRING_INTERNER.intern(dataSource);
     this.interval = interval;
     this.loadSpec = loadSpec;
     this.version = version;
+
     this.dimensions = dimensions == null
                       ? ImmutableList.<String>of()
-                      : ImmutableList.copyOf(Iterables.transform(Iterables.filter(dimensions, nonEmpty), internFun));
+                      : LIST_INTERNER.intern(Ordering.natural()
+                                                     .immutableSortedCopy(Iterables.filter(dimensions, nonEmpty)));
     this.metrics = metrics == null
                    ? ImmutableList.<String>of()
-                   : ImmutableList.copyOf(Iterables.transform(Iterables.filter(metrics, nonEmpty), internFun));
-    this.shardSpec = (shardSpec == null) ? new NoneShardSpec() : shardSpec;
+                   : LIST_INTERNER.intern(Ordering.natural().immutableSortedCopy(Iterables.filter(metrics, nonEmpty)));
+    this.shardSpec = (shardSpec == null) ? NoneShardSpec.INSTANCE : shardSpec;
     this.binaryVersion = binaryVersion;
     this.size = size;
 
@@ -139,6 +136,7 @@ public boolean apply(String input)
     );
   }
 
+
   /**
    * Get dataSource
    *
@@ -323,7 +321,7 @@ public Builder()
       this.loadSpec = ImmutableMap.of();
       this.dimensions = ImmutableList.of();
       this.metrics = ImmutableList.of();
-      this.shardSpec = new NoneShardSpec();
+      this.shardSpec = NoneShardSpec.INSTANCE;
       this.size = -1;
     }
 

api/src/main/java/io/druid/timeline/partition/NoneShardSpec.java

@@ -19,6 +19,7 @@
 
 package io.druid.timeline.partition;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Iterables;
 import com.metamx.common.ISE;
@@ -32,6 +33,11 @@
  */
 public class NoneShardSpec implements ShardSpec
 {
+  public final static NoneShardSpec INSTANCE = new NoneShardSpec();
+
+  @JsonCreator
+  public static NoneShardSpec instance() { return INSTANCE; }
+
   @Override
   public <T> PartitionChunk<T> createChunk(T obj)
   {

api/src/test/java/io/druid/timeline/partition/NoneShardSpecTest.java

@@ -1,5 +1,8 @@
 package io.druid.timeline.partition;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.druid.TestObjectMapper;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -13,4 +16,17 @@ public void testEqualsAndHashCode()
     Assert.assertEquals(one, two);
     Assert.assertEquals(one.hashCode(), two.hashCode());
   }
+
+  @Test
+  public void testSerde() throws Exception
+  {
+    final NoneShardSpec one = NoneShardSpec.INSTANCE;
+    ObjectMapper mapper = new TestObjectMapper();
+    NoneShardSpec serde1 = mapper.readValue(mapper.writeValueAsString(one), NoneShardSpec.class);
+    NoneShardSpec serde2 = mapper.readValue(mapper.writeValueAsString(one), NoneShardSpec.class);
+
+    // Serde should return same object instead of creating new one every time.
+    Assert.assertTrue(serde1 == serde2);
+    Assert.assertTrue(one == serde1);
+  }
 }

indexing-hadoop/src/test/java/io/druid/indexer/BatchDeltaIngestionTest.java

@@ -305,8 +305,8 @@ private void testIngestion(
     Assert.assertEquals("local", dataSegment.getLoadSpec().get("type"));
     Assert.assertEquals(indexZip.getCanonicalPath(), dataSegment.getLoadSpec().get("path"));
     Assert.assertEquals("host", dataSegment.getDimensions().get(0));
-    Assert.assertEquals("visited_sum", dataSegment.getMetrics().get(0));
-    Assert.assertEquals("unique_hosts", dataSegment.getMetrics().get(1));
+    Assert.assertEquals("unique_hosts", dataSegment.getMetrics().get(0));
+    Assert.assertEquals("visited_sum", dataSegment.getMetrics().get(1));
     Assert.assertEquals(Integer.valueOf(9), dataSegment.getBinaryVersion());
 
     HashBasedNumberedShardSpec spec = (HashBasedNumberedShardSpec) dataSegment.getShardSpec();

indexing-hadoop/src/test/java/io/druid/indexer/IndexGeneratorJobTest.java

@@ -597,11 +597,11 @@ private void verifyJob(IndexGeneratorJob job) throws IOException
         if (datasourceName.equals("website")) {
           Assert.assertEquals("website", dataSegment.getDataSource());
           Assert.assertEquals("host", dataSegment.getDimensions().get(0));
-          Assert.assertEquals("visited_num", dataSegment.getMetrics().get(0));
-          Assert.assertEquals("unique_hosts", dataSegment.getMetrics().get(1));
+          Assert.assertEquals("unique_hosts", dataSegment.getMetrics().get(0));
+          Assert.assertEquals("visited_num", dataSegment.getMetrics().get(1));
         } else if (datasourceName.equals("inherit_dims")) {
           Assert.assertEquals("inherit_dims", dataSegment.getDataSource());
-          Assert.assertEquals(ImmutableList.of("X", "Y", "M", "Q", "B", "F"), dataSegment.getDimensions());
+          Assert.assertEquals(ImmutableList.of("B", "F", "M", "Q", "X", "Y"), dataSegment.getDimensions());
           Assert.assertEquals("count", dataSegment.getMetrics().get(0));
         } else if (datasourceName.equals("inherit_dims2")) {
           Assert.assertEquals("inherit_dims2", dataSegment.getDataSource());