addressed comments

docs/content/querying/filters.md

@@ -147,4 +147,13 @@ Search filters can be used to filter on partial string matches.
 |property|description|required?|
 |--------|-----------|---------|
 |type|This String should always be "fragment".|yes|
-|values|A JSON array of String values to run the search over. Case insensitive.|yes|
+|values|A JSON array of String values to run the search over.|yes|
+|caseSensitive|Whether strings should be compared as case sensitive or not. Default: false(insensitive)|no|
+
+##### Contains
+
+|property|description|required?|
+|--------|-----------|---------|
+|type|This String should always be "contains".|yes|
+|value|A String value to run the search over.|yes|
+|caseSensitive|Whether two string should be compared as case sensitive or not|yes|

docs/content/querying/searchqueryspec.md

@@ -19,11 +19,25 @@ If any part of a dimension value contains the value specified in this search que
 FragmentSearchQuerySpec
 -----------------------
 
-If any part of a dimension value contains any of the values specified in this search query spec, regardless of case, a "match" occurs. The grammar is:
+If any part of a dimension value contains all of the values specified in this search query spec, regardless of case by default, a "match" occurs. The grammar is:
 
 ```json
 { 
   "type" : "fragment",
+  "case_sensitive" : false,
   "values" : ["fragment1", "fragment2"]
 }
 ```
+
+ContainsSearchQuerySpec
+----------------------------------
+
+If any part of a dimension value contains the value specified in this search query spec, a "match" occurs. The grammar is:
+
+```json
+{
+  "type"  : "contains",
+  "case_sensitive" : true,
+  "value" : "some_value"
+}
+```

processing/pom.xml

@@ -75,6 +75,10 @@
             <groupId>org.mapdb</groupId>
             <artifactId>mapdb</artifactId>
         </dependency>
+        <dependency>
+            <groupId>commons-lang</groupId>
+            <artifactId>commons-lang</artifactId>
+        </dependency>
 
         <!-- Tests -->
         <dependency>

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

@@ -18,6 +18,7 @@
 package io.druid.query;
 
 import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
@@ -683,25 +684,29 @@ public class Druids
 
     public SearchQueryBuilder query(String q)
     {
+      Preconditions.checkNotNull(q, "no value");
       querySpec = new InsensitiveContainsSearchQuerySpec(q);
       return this;
     }
 
     public SearchQueryBuilder query(Map<String, Object> q)
     {
-      querySpec = new InsensitiveContainsSearchQuerySpec((String) q.get("value"));
+      String value = Preconditions.checkNotNull(q.get("value"), "no value").toString();
+      querySpec = new InsensitiveContainsSearchQuerySpec(value);
       return this;
     }
 
     public SearchQueryBuilder query(String q, boolean caseSensitive)
     {
+      Preconditions.checkNotNull(q, "no value");
       querySpec = new ContainsSearchQuerySpec(q, caseSensitive);
       return this;
     }
 
     public SearchQueryBuilder query(Map<String, Object> q, boolean caseSensitive)
     {
-      querySpec = new ContainsSearchQuerySpec((String) q.get("value"), caseSensitive);
+      String value = Preconditions.checkNotNull(q.get("value"), "no value").toString();
+      querySpec = new ContainsSearchQuerySpec(value, caseSensitive);
       return this;
     }
 
@@ -712,6 +717,7 @@ public class Druids
 
     public SearchQueryBuilder fragments(List<String> q, boolean caseSensitive)
     {
+      Preconditions.checkNotNull(q, "no value");
       querySpec = new FragmentSearchQuerySpec(q, caseSensitive);
       return this;
     }

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

@@ -1,3 +1,20 @@
+/*
+ * Druid - a distributed column store.
+ * Copyright 2012 - 2015 Metamarkets Group Inc.
+ *
+ * Licensed 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 io.druid.query.search.search;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -16,8 +33,6 @@ public class ContainsSearchQuerySpec implements SearchQuerySpec
   private final String value;
   private final boolean caseSensitive;
 
-  private final String target;
-
   @JsonCreator
   public ContainsSearchQuerySpec(
       @JsonProperty("value") String value,
@@ -26,7 +41,6 @@ public class ContainsSearchQuerySpec implements SearchQuerySpec
   {
     this.value = value;
     this.caseSensitive = caseSensitive;
-    this.target = value == null || caseSensitive ? value : value.toLowerCase();
   }
 
   @JsonProperty
@@ -44,21 +58,29 @@ public class ContainsSearchQuerySpec implements SearchQuerySpec
   @Override
   public boolean accept(String dimVal)
   {
-    if (dimVal == null) {
+    if (dimVal == null || value == null) {
       return false;
     }
-    final String input = caseSensitive ? dimVal : dimVal.toLowerCase();
+    if (caseSensitive) {
-    return input.contains(target);
+      return dimVal.contains(value);
+    }
+    return org.apache.commons.lang.StringUtils.containsIgnoreCase(dimVal, value);
   }
 
   @Override
   public byte[] getCacheKey()
   {
+    if (value == null) {
+      return ByteBuffer.allocate(2)
+                       .put(CACHE_TYPE_ID)
+                       .put(caseSensitive ? (byte) 1 : 0).array();
+    }
+
     byte[] valueBytes = StringUtils.toUtf8(value);
 
     return ByteBuffer.allocate(2 + valueBytes.length)
-                     .put(caseSensitive ? (byte)1 : 0)
                      .put(CACHE_TYPE_ID)
+                     .put(caseSensitive ? (byte) 1 : 0)
                      .put(valueBytes)
                      .array();
   }

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

@@ -40,7 +40,8 @@ public class FragmentSearchQuerySpec implements SearchQuerySpec
   @JsonCreator
   public FragmentSearchQuerySpec(
       @JsonProperty("values") List<String> values
-  ) {
+  )
+  {
     this(values, false);
   }
 
@@ -55,7 +56,7 @@ public class FragmentSearchQuerySpec implements SearchQuerySpec
     Set<String> set = new HashSet();
     if (values != null) {
       for (String value : values) {
-        set.add(caseSensitive ? value : value.toLowerCase());
+        set.add(value);
       }
     }
     target = set.toArray(new String[set.size()]);
@@ -76,10 +77,22 @@ public class FragmentSearchQuerySpec implements SearchQuerySpec
   @Override
   public boolean accept(String dimVal)
   {
-    if (dimVal == null) {
+    if (dimVal == null || values == null) {
       return false;
     }
-    final String input = caseSensitive ? dimVal : dimVal.toLowerCase();
+    if (caseSensitive) {
+      return containsAny(target, dimVal);
+    }
+    for (String search : target) {
+      if (!org.apache.commons.lang.StringUtils.containsIgnoreCase(dimVal, search)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private boolean containsAny(String[] target, String input)
+  {
     for (String value : target) {
       if (!input.contains(value)) {
         return false;
@@ -91,6 +104,12 @@ public class FragmentSearchQuerySpec implements SearchQuerySpec
   @Override
   public byte[] getCacheKey()
   {
+    if (values == null) {
+      return ByteBuffer.allocate(2)
+                       .put(CACHE_TYPE_ID)
+                       .put(caseSensitive ? (byte) 1 : 0).array();
+    }
+
     final byte[][] valuesBytes = new byte[values.size()][];
     int valuesBytesSize = 0;
     int index = 0;
@@ -101,8 +120,8 @@ public class FragmentSearchQuerySpec implements SearchQuerySpec
     }
 
     final ByteBuffer queryCacheKey = ByteBuffer.allocate(2 + valuesBytesSize)
-        .put(caseSensitive ? (byte) 1 : 0)
+                                               .put(CACHE_TYPE_ID)
-        .put(CACHE_TYPE_ID);
+                                               .put(caseSensitive ? (byte) 1 : 0);
 
     for (byte[] bytes : valuesBytes) {
       queryCacheKey.put(bytes);

processing/src/test/java/io/druid/query/QueryRunnerTestHelper.java

@@ -329,10 +329,19 @@ public class QueryRunnerTestHelper
       QueryRunnerFactory<T, QueryType> factory,
       Segment adapter
   )
+  {
+    return makeQueryRunner(factory, segmentId, adapter);
+  }
+
+  public static <T, QueryType extends Query<T>> QueryRunner<T> makeQueryRunner(
+      QueryRunnerFactory<T, QueryType> factory,
+      String segmentId,
+      Segment adapter
+  )
   {
     return new FinalizeResultsQueryRunner<T>(
         new BySegmentQueryRunner<T>(
-            adapter.getIdentifier(), adapter.getDataInterval().getStart(),
+            segmentId, adapter.getDataInterval().getStart(),
             factory.createRunner(adapter)
         ),
         (QueryToolChest<T, Query<T>>)factory.getToolchest()

processing/src/test/java/io/druid/query/search/SearchQueryRunnerWithCaseTest.java

@@ -68,7 +68,8 @@ public class SearchQueryRunnerWithCaseTest
         "2011-01-12T00:00:00.000Z\tspot\tAutoMotive\tPREFERRED\ta\u0001preferred\t100.000000\n" +
         "2011-01-12T00:00:00.000Z\tSPot\tbusiness\tpreferred\tb\u0001Preferred\t100.000000\n" +
         "2011-01-12T00:00:00.000Z\tspot\tentertainment\tPREFERRed\te\u0001preferred\t100.000000\n" +
-        "2011-01-13T00:00:00.000Z\tspot\tautomotive\tpreferred\ta\u0001preferred\t94.874713");
+        "2011-01-13T00:00:00.000Z\tspot\tautomotive\tpreferred\ta\u0001preferred\t94.874713"
+    );
 
     IncrementalIndex index1 = TestIndex.makeRealtimeIndex(input, true);
     IncrementalIndex index2 = TestIndex.makeRealtimeIndex(input, false);
@@ -76,12 +77,14 @@ public class SearchQueryRunnerWithCaseTest
     QueryableIndex index3 = TestIndex.persistRealtimeAndLoadMMapped(index1);
     QueryableIndex index4 = TestIndex.persistRealtimeAndLoadMMapped(index2);
 
-    return transformToConstructionFeeder(Arrays.asList(
+    return transformToConstructionFeeder(
-        makeQueryRunner(factory, new IncrementalIndexSegment(index1, "index1")),
+        Arrays.asList(
-        makeQueryRunner(factory, new IncrementalIndexSegment(index2, "index2")),
+            makeQueryRunner(factory, "index1", new IncrementalIndexSegment(index1, "index1")),
-        makeQueryRunner(factory, new QueryableIndexSegment("index3", index3)),
+            makeQueryRunner(factory, "index2", new IncrementalIndexSegment(index2, "index2")),
-        makeQueryRunner(factory, new QueryableIndexSegment("index4", index4))
+            makeQueryRunner(factory, "index3", new QueryableIndexSegment("index3", index3)),
-    ));
+            makeQueryRunner(factory, "index4", new QueryableIndexSegment("index4", index4))
+        )
+    );
   }
 
   private final QueryRunner runner;
@@ -93,7 +96,8 @@ public class SearchQueryRunnerWithCaseTest
     this.runner = runner;
   }
 
-  private Druids.SearchQueryBuilder testBuilder() {
+  private Druids.SearchQueryBuilder testBuilder()
+  {
     return Druids.newSearchQueryBuilder()
                  .dataSource(dataSource)
                  .granularity(allGran)

processing/src/test/java/io/druid/segment/TestIndex.java

@@ -162,7 +162,8 @@ public class TestIndex
     }
   }
 
-  private static IncrementalIndex makeRealtimeIndex(final String resourceFilename, final boolean useOffheap) {
+  private static IncrementalIndex makeRealtimeIndex(final String resourceFilename, final boolean useOffheap)
+  {
     final URL resource = TestIndex.class.getClassLoader().getResource(resourceFilename);
     if (resource == null) {
       throw new IllegalArgumentException("cannot find resource " + resourceFilename);
@@ -198,7 +199,8 @@ public class TestIndex
     int lineCount;
     try {
       lineCount = source.readLines(
-          new LineProcessor<Integer>() {
+          new LineProcessor<Integer>()
+          {
             StringInputRowParser parser = new StringInputRowParser(
                 new DelimitedParseSpec(
                     new TimestampSpec("ts", "iso", null),