Synchronization of lookups during startup of druid processes

docs/content/querying/lookups.md

@@ -323,6 +323,8 @@ It is possible to save the configuration across restarts such that a node will n
 |Property|Description|Default|
 |--------|-----------|-------|
 |`druid.lookup.snapshotWorkingDir`| Working path used to store snapshot of current lookup configuration, leaving this property null will disable snapshot/bootstrap utility|null|
+|`druid.lookup.numLookupLoadingThreads`| Number of threads for loading the lookups in parallel on startup. This thread pool is destroyed once startup is done. It is not kept during the lifetime of the JVM|Available Processors / 2|
+|`druid.lookup.enableLookupSyncOnStartup`|Enable the lookup synchronization process with coordinator on startup. The queryable nodes will fetch and load the lookups from the coordinator instead of waiting for the coordinator to load the lookups for them. Users may opt to disable this option if there are no lookups configured in the cluster.|true|
 
 ## Introspect a Lookup
 

extensions-core/histogram/pom.xml

@@ -59,6 +59,13 @@
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-server</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

indexing-service/src/test/java/io/druid/indexing/common/TestUtils.java

@@ -29,7 +29,7 @@
 import io.druid.java.util.common.ISE;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.math.expr.ExprMacroTable;
-import io.druid.query.expression.TestExprMacroTable;
+import io.druid.query.expression.TestExpressionMacroTable;
 import io.druid.segment.IndexIO;
 import io.druid.segment.IndexMergerV9;
 import io.druid.segment.column.ColumnConfig;
@@ -74,7 +74,7 @@ public int columnCacheSizeBytes()
 
     jsonMapper.setInjectableValues(
         new InjectableValues.Std()
-            .addValue(ExprMacroTable.class.getName(), TestExprMacroTable.INSTANCE)
+            .addValue(ExprMacroTable.class.getName(), TestExpressionMacroTable.INSTANCE)
             .addValue(IndexIO.class, indexIO)
             .addValue(ObjectMapper.class, jsonMapper)
             .addValue(ChatHandlerProvider.class, new NoopChatHandlerProvider())

processing/src/main/java/io/druid/query/dimension/DimensionSpec.java

@@ -35,8 +35,7 @@
     @JsonSubTypes.Type(name = "default", value = DefaultDimensionSpec.class),
     @JsonSubTypes.Type(name = "extraction", value = ExtractionDimensionSpec.class),
     @JsonSubTypes.Type(name = "regexFiltered", value = RegexFilteredDimensionSpec.class),
-    @JsonSubTypes.Type(name = "listFiltered", value = ListFilteredDimensionSpec.class),
-    @JsonSubTypes.Type(name = "lookup", value = LookupDimensionSpec.class)
+    @JsonSubTypes.Type(name = "listFiltered", value = ListFilteredDimensionSpec.class)
 })
 public interface DimensionSpec extends Cacheable
 {

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

@@ -24,7 +24,6 @@
 import io.druid.guice.annotations.ExtensionPoint;
 import io.druid.java.util.common.Cacheable;
 import io.druid.query.lookup.LookupExtractionFn;
-import io.druid.query.lookup.RegisteredLookupExtractionFn;
 
 import javax.annotation.Nullable;
 
@@ -41,7 +40,6 @@
     @JsonSubTypes.Type(name = "timeFormat", value = TimeFormatExtractionFn.class),
     @JsonSubTypes.Type(name = "identity", value = IdentityExtractionFn.class),
     @JsonSubTypes.Type(name = "lookup", value = LookupExtractionFn.class),
-    @JsonSubTypes.Type(name = "registeredLookup", value = RegisteredLookupExtractionFn.class),
     @JsonSubTypes.Type(name = "substring", value = SubstringDimExtractionFn.class),
     @JsonSubTypes.Type(name = "cascade", value = CascadeExtractionFn.class),
     @JsonSubTypes.Type(name = "stringFormat", value = StringFormatExtractionFn.class),

processing/src/main/java/io/druid/query/lookup/LookupConfig.java

@@ -26,11 +26,19 @@
 public class LookupConfig
 {
 
-  @JsonProperty
-  private final String snapshotWorkingDir;
+  @JsonProperty("snapshotWorkingDir")
+  private String snapshotWorkingDir;
+
+  @JsonProperty("enableLookupSyncOnStartup")
+  private boolean enableLookupSyncOnStartup = true;
+
+  @JsonProperty("numLookupLoadingThreads")
+  private int numLookupLoadingThreads = Runtime.getRuntime().availableProcessors() / 2;
 
   /**
-   * @param snapshotWorkingDir working directory to store lookups snapshot file, passing null or empty string will disable the snapshot utility
+   * @param snapshotWorkingDir        working directory to store lookups snapshot file, passing null or empty string will disable the snapshot utility
+   * @param numLookupLoadingThreads   number of threads for loading the lookups as part of the synchronization process
+   * @param enableLookupSyncOnStartup decides whether the lookup synchronization process should be enabled at startup
    */
   @JsonCreator
   public LookupConfig(
@@ -45,6 +53,15 @@ public String getSnapshotWorkingDir()
     return snapshotWorkingDir;
   }
 
+  public int getNumLookupLoadingThreads()
+  {
+    return numLookupLoadingThreads;
+  }
+
+  public boolean getEnableLookupSyncOnStartup()
+  {
+    return enableLookupSyncOnStartup;
+  }
 
   @Override
   public boolean equals(Object o)
@@ -58,7 +75,9 @@ public boolean equals(Object o)
 
     LookupConfig that = (LookupConfig) o;
 
-    return getSnapshotWorkingDir().equals(that.getSnapshotWorkingDir());
+    return snapshotWorkingDir.equals(that.snapshotWorkingDir) &&
+           enableLookupSyncOnStartup == that.enableLookupSyncOnStartup &&
+           numLookupLoadingThreads == that.numLookupLoadingThreads;
 
   }
 
@@ -67,6 +86,8 @@ public String toString()
   {
     return "LookupConfig{" +
            "snapshotWorkingDir='" + getSnapshotWorkingDir() + '\'' +
+           " numLookupLoadingThreads='" + getNumLookupLoadingThreads() + '\'' +
+           " enableLookupSyncOnStartup='" + getEnableLookupSyncOnStartup() + '\'' +
            '}';
   }
 }

processing/src/test/java/io/druid/query/expression/TestExprMacroTable.java

@@ -20,17 +20,7 @@
 package io.druid.query.expression;
 
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import io.druid.math.expr.ExprMacroTable;
-import io.druid.query.extraction.MapLookupExtractor;
-import io.druid.query.lookup.LookupExtractor;
-import io.druid.query.lookup.LookupExtractorFactory;
-import io.druid.query.lookup.LookupExtractorFactoryContainer;
-import io.druid.query.lookup.LookupIntrospectHandler;
-import io.druid.query.lookup.LookupReferencesManager;
-import org.easymock.EasyMock;
-
-import javax.annotation.Nullable;
 
 public class TestExprMacroTable extends ExprMacroTable
 {
@@ -41,7 +31,6 @@ private TestExprMacroTable()
     super(
         ImmutableList.of(
             new LikeExprMacro(),
-            new LookupExprMacro(createTestLookupReferencesManager(ImmutableMap.of("foo", "xfoo"))),
             new RegexpExtractExprMacro(),
             new TimestampCeilExprMacro(),
             new TimestampExtractExprMacro(),
@@ -55,52 +44,4 @@ private TestExprMacroTable()
         )
     );
   }
-
-  /**
-   * Returns a mock {@link LookupReferencesManager} that has one lookup, "lookyloo".
-   */
-  public static LookupReferencesManager createTestLookupReferencesManager(final ImmutableMap<String, String> theLookup)
-  {
-    final LookupReferencesManager lookupReferencesManager = EasyMock.createMock(LookupReferencesManager.class);
-    EasyMock.expect(lookupReferencesManager.get(EasyMock.eq("lookyloo"))).andReturn(
-        new LookupExtractorFactoryContainer(
-            "v0",
-            new LookupExtractorFactory()
-            {
-              @Override
-              public boolean start()
-              {
-                throw new UnsupportedOperationException();
-              }
-
-              @Override
-              public boolean close()
-              {
-                throw new UnsupportedOperationException();
-              }
-
-              @Override
-              public boolean replaces(@Nullable final LookupExtractorFactory other)
-              {
-                throw new UnsupportedOperationException();
-              }
-
-              @Override
-              public LookupIntrospectHandler getIntrospectHandler()
-              {
-                throw new UnsupportedOperationException();
-              }
-
-              @Override
-              public LookupExtractor get()
-              {
-                return new MapLookupExtractor(theLookup, false);
-              }
-            }
-        )
-    ).anyTimes();
-    EasyMock.expect(lookupReferencesManager.get(EasyMock.not(EasyMock.eq("lookyloo")))).andReturn(null).anyTimes();
-    EasyMock.replay(lookupReferencesManager);
-    return lookupReferencesManager;
-  }
 }

processing/src/test/java/io/druid/query/lookup/LookupConfigTest.java

@@ -32,13 +32,36 @@ public class LookupConfigTest
 {
 
   ObjectMapper mapper = TestHelper.getJsonMapper();
-
   @Rule
   public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
   @Test
   public void TestSerDesr() throws IOException
   {
     LookupConfig lookupConfig = new LookupConfig(temporaryFolder.newFile().getAbsolutePath());
-    Assert.assertEquals(lookupConfig, mapper.reader(LookupConfig.class).readValue(mapper.writeValueAsString(lookupConfig)));
+    Assert.assertEquals(
+        lookupConfig,
+        mapper.reader(LookupConfig.class).readValue(mapper.writeValueAsString(lookupConfig))
+    );
+  }
+
+  @Test
+  public void testSerdeWithNonDefaults() throws Exception
+  {
+    String json = "{\n"
+                  + "  \"enableLookupSyncOnStartup\": false,\n"
+                  + "  \"snapshotWorkingDir\": \"/tmp\",\n"
+                  + "  \"numLookupLoadingThreads\": 4 \n"
+                  + "}\n";
+    LookupConfig config = mapper.readValue(
+        mapper.writeValueAsString(
+            mapper.readValue(json, LookupConfig.class)
+        ),
+        LookupConfig.class
+    );
+
+    Assert.assertEquals("/tmp", config.getSnapshotWorkingDir());
+    Assert.assertEquals(false, config.getEnableLookupSyncOnStartup());
+    Assert.assertEquals(4, config.getNumLookupLoadingThreads());
   }
 }

server/src/main/java/io/druid/query/lookup/LookupModule.java

@@ -25,6 +25,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.Module;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
@@ -48,6 +49,7 @@
 import io.druid.guice.annotations.Smile;
 import io.druid.initialization.DruidModule;
 import io.druid.java.util.common.logger.Logger;
+import io.druid.query.dimension.LookupDimensionSpec;
 import io.druid.query.expression.LookupExprMacro;
 import io.druid.server.DruidNode;
 import io.druid.server.http.HostAndPortWithScheme;
@@ -84,7 +86,11 @@ public static String getTierListenerPath(String tier)
   public List<? extends Module> getJacksonModules()
   {
     return ImmutableList.<Module>of(
-        new SimpleModule("DruidLookupModule").registerSubtypes(MapLookupExtractorFactory.class)
+        new SimpleModule("DruidLookupModule").registerSubtypes(MapLookupExtractorFactory.class),
+        new SimpleModule().registerSubtypes(
+            new NamedType(LookupDimensionSpec.class, "lookup"),
+            new NamedType(RegisteredLookupExtractionFn.class, "registeredLookup")
+        )
     );
   }