adding bulk lookup and reverse lookup

extensions/namespace-lookup/src/main/java/io/druid/query/extraction/NamespacedExtractor.java

@@ -37,7 +37,7 @@ import java.util.List;
  * In the event that an unknown namespace is passed, a simple reflective function is returned instead.
  */
 @JsonTypeName("namespace")
-public class NamespacedExtractor implements LookupExtractor
+public class NamespacedExtractor extends LookupExtractor
 {
   private static final byte CACHE_TYPE_ID = 0x05;
 
@@ -83,7 +83,7 @@ public class NamespacedExtractor implements LookupExtractor
   }
 
   @Override
-  public List<String> unApply(@NotNull String value)
+  public List<String> unapply(@NotNull String value)
   {
     return reverseExtractionFunction.apply(value);
   }

extensions/namespace-lookup/src/test/java/io/druid/query/extraction/namespace/NamespacedExtractorTest.java

@@ -137,11 +137,11 @@ public class NamespacedExtractorTest
     for (int i = 0; i < 10; ++i) {
       final String val = UUID.randomUUID().toString();
       Assert.assertEquals(val, namespacedExtractor.apply(val));
-      Assert.assertEquals(Arrays.asList(val), namespacedExtractor.unApply(val));
+      Assert.assertEquals(Arrays.asList(val), namespacedExtractor.unapply(val));
     }
     Assert.assertEquals("", namespacedExtractor.apply(""));
     Assert.assertNull(namespacedExtractor.apply(null));
-    Assert.assertEquals(Collections.emptyList(), namespacedExtractor.unApply(null));
+    Assert.assertEquals(Collections.emptyList(), namespacedExtractor.unapply(null));
     Assert.assertEquals("The awesomeness", namespacedExtractor.apply("The awesomeness"));
   }
 

processing/src/main/java/io/druid/query/extraction/LookupExtractor.java

@@ -25,23 +25,50 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo;
 
 import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes(value = {
     @JsonSubTypes.Type(name = "map", value = MapLookupExtractor.class)
 })
-public interface LookupExtractor
+public abstract class LookupExtractor
 {
   /**
    * Apply a particular lookup methodology to the input string
+   *
    * @param key The value to apply the lookup to. May not be null
+   *
    * @return The lookup, or null key cannot have the lookup applied to it and should be treated as missing.
    */
-  @Nullable String apply(@NotNull String key);
+  @Nullable
+  abstract String apply(@NotNull String key);
+
+  /**
+   * @param keys set of keys to apply lookup for each element
+   *
+   * @return Returns {@link Map} whose keys are the contents of {@code keys} and whose values are computed on demand using lookup function {@link #unapply(String)}
+   * or empty map if {@code values} is `null`
+   * User can override this method if there is a better way to perform bulk lookup
+   */
+
+  Map<String, String> applyAll(Iterable<String> keys)
+  {
+    if (keys == null) {
+      return Collections.emptyMap();
+    }
+    Map<String, String> map = new HashMap<>();
+    for (String key : keys) {
+      map.put(key, apply(key));
+    }
+    return map;
+  }
 
   /**
    * Provide the reverse mapping from a given value to a list of keys
+   *
    * @param value the value to apply the reverse lookup
    *              Null and empty are considered to be the same value = nullToEmpty(value)
    *
@@ -50,11 +77,35 @@ public interface LookupExtractor
    * returning an empty list implies that user want to ignore such a lookup value.
    * In the other hand returning a list with the null element implies user want to map the none existing value to the key null.
    */
-  List<String> unApply(String value);
+
+  abstract List<String> unapply(String value);
+
+  /**
+   * @param values Iterable of values for which will perform reverse lookup
+   *
+   * @return Returns {@link Map} whose keys are the contents of {@code values} and whose values are computed on demand using the reverse lookup function {@link #unapply(String)}
+   * or empty map if {@code values} is `null`
+   * User can override this method if there is a better way to perform bulk reverse lookup
+   */
+
+  Map<String, List<String>> unapplyAll(Iterable<String> values)
+  {
+    if (values == null) {
+      return Collections.emptyMap();
+    }
+    Map<String, List<String>> map = new HashMap<>();
+    for (String value : values) {
+      map.put(value, unapply(value));
+    }
+    return map;
+  }
 
   /**
    * Create a cache key for use in results caching
+   *
    * @return A byte array that can be used to uniquely identify if results of a prior lookup can use the cached values
    */
-  @NotNull byte[] getCacheKey();
+
+  @Nullable
+  abstract byte[] getCacheKey();
 }

processing/src/main/java/io/druid/query/extraction/MapLookupExtractor.java

@@ -39,7 +39,7 @@ import java.util.List;
 import java.util.Map;
 
 @JsonTypeName("map")
-public class MapLookupExtractor implements LookupExtractor
+public class MapLookupExtractor extends LookupExtractor
 {
   private final Map<String, String> map;
 
@@ -65,7 +65,7 @@ public class MapLookupExtractor implements LookupExtractor
   }
 
   @Override
-  public List<String> unApply(final String value)
+  public List<String> unapply(final String value)
   {
     return Lists.newArrayList(Maps.filterKeys(map, new Predicate<String>()
     {

processing/src/test/java/io/druid/query/extraction/LookupExtractorTest.java

@@ -0,0 +1,102 @@
+/*
+ *
+ *  Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ *  or more contributor license agreements. See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership. Metamarkets licenses this file
+ *  to you 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.extraction;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class LookupExtractorTest
+{
+
+  static final Map<String, String> EXPECTED_MAP = ImmutableMap.of(
+      "key1",
+      "value1",
+      "key2",
+      "value2",
+      "key-1",
+      "value1",
+      "",
+      "emptyString"
+  );
+
+  static final Map<String, List<String>> EXPECTED_REVERSE_MAP = ImmutableMap.of(
+      "value1",
+      Arrays.asList("key1", "key-1"),
+      "value2",
+      Arrays.asList("key2"),
+      "emptyString",
+      Arrays.asList("")
+  );
+  LookupExtractor lookupExtractor = new MapLookupExtractor(EXPECTED_MAP);
+
+  @Test
+  public void testApplyAll()
+  {
+    Assert.assertEquals(EXPECTED_MAP, lookupExtractor.applyAll(EXPECTED_MAP.keySet()));
+  }
+
+  @Test
+  public void testApplyAllWithNull()
+  {
+    Assert.assertEquals(Collections.emptyMap(), lookupExtractor.applyAll(null));
+  }
+
+  @Test
+  public void testApplyAllWithEmptySet()
+  {
+    Assert.assertEquals(Collections.emptyMap(), lookupExtractor.applyAll(Collections.<String>emptySet()));
+  }
+
+  @Test
+  public void testApplyAllWithNotExisting()
+  {
+    Map<String, String> actual = new HashMap<>();
+    actual.put("not there", null);
+    Assert.assertEquals(actual, lookupExtractor.applyAll(Lists.newArrayList("not there")));
+  }
+
+  @Test
+  public void testUnapplyAllWithNull()
+  {
+    Assert.assertEquals(Collections.emptyMap(), lookupExtractor.unapplyAll(null));
+  }
+
+  @Test
+  public void testunapplyAllWithEmptySet()
+  {
+    Assert.assertEquals(Collections.emptyMap(), lookupExtractor.unapplyAll(Collections.<String>emptySet()));
+  }
+
+  @Test
+  public void testUnapplyAll()
+  {
+    Assert.assertEquals(EXPECTED_REVERSE_MAP, lookupExtractor.unapplyAll(EXPECTED_MAP.values()));
+  }
+}

processing/src/test/java/io/druid/query/extraction/MapLookupExtractorTest.java

@@ -37,13 +37,13 @@ public class MapLookupExtractorTest
   @Test
   public void testUnApply()
   {
-    Assert.assertEquals(Arrays.asList("foo"), fn.unApply("bar"));
-    Assert.assertEquals(Sets.newHashSet("null", "empty String"), Sets.newHashSet(fn.unApply("")));
+    Assert.assertEquals(Arrays.asList("foo"), fn.unapply("bar"));
+    Assert.assertEquals(Sets.newHashSet("null", "empty String"), Sets.newHashSet(fn.unapply("")));
     Assert.assertEquals("Null value should be equal to empty string",
                         Sets.newHashSet("null", "empty String"),
-                        Sets.newHashSet(fn.unApply(null)));
-    Assert.assertEquals(Sets.newHashSet(""), Sets.newHashSet(fn.unApply("empty_string")));
-    Assert.assertEquals("not existing value returns empty list", Collections.EMPTY_LIST, fn.unApply("not There"));
+                        Sets.newHashSet(fn.unapply((String) null)));
+    Assert.assertEquals(Sets.newHashSet(""), Sets.newHashSet(fn.unapply("empty_string")));
+    Assert.assertEquals("not existing value returns empty list", Collections.EMPTY_LIST, fn.unapply("not There"));
   }
 
   @Test