Add JoinableFactory interface and use it in the query stack. (#9247)

* Add JoinableFactory interface and use it in the query stack. Also includes InlineJoinableFactory, which enables joining against inline datasources. This is the first patch where a basic join query actually works. It includes integration tests. * Fix test issues. * Adjustments from code review.
apache · Jan 24, 2020 · 19b427e · 19b427e
1 parent 3daf0f8
commit 19b427e
Show file tree

Hide file tree

Showing 63 changed files with 1,120 additions and 96 deletions.
diff --git a/core/src/main/java/org/apache/druid/utils/JvmUtils.java b/core/src/main/java/org/apache/druid/utils/JvmUtils.java
@@ -29,6 +29,8 @@
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.StringTokenizer;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -80,13 +82,29 @@ public static long safeGetThreadCpuTime()
    * @return total CPU time for the current thread in nanoseconds.
    *
    * @throws UnsupportedOperationException if the Java virtual machine does not support CPU time measurement for
-   * the current thread.
+   *                                       the current thread.
    */
   public static long getCurrentThreadCpuTime()
   {
     return THREAD_MX_BEAN.getCurrentThreadCpuTime();
   }
 
+  /**
+   * Executes and returns the value of {@code function}. Also accumulates the CPU time taken for the function (as
+   * reported by {@link #getCurrentThreadCpuTime()} into {@param accumulator}.
+   */
+  public static <T> T safeAccumulateThreadCpuTime(final AtomicLong accumulator, final Supplier<T> function)
+  {
+    final long start = safeGetThreadCpuTime();
+
+    try {
+      return function.get();
+    }
+    finally {
+      accumulator.addAndGet(safeGetThreadCpuTime() - start);
+    }
+  }
+
   public static List<URL> systemClassPath()
   {
     List<URL> jobURLs;

diff --git a/...ka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java b/...ka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
@@ -105,6 +105,7 @@
 import org.apache.druid.query.timeseries.TimeseriesQueryQueryToolChest;
 import org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory;
 import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.DataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
@@ -2652,6 +2653,7 @@ public void close()
         handoffNotifierFactory,
         this::makeTimeseriesAndScanConglomerate,
         Execs.directExecutor(), // queryExecutorService
+        NoopJoinableFactory.INSTANCE,
         EasyMock.createMock(MonitorScheduler.class),
         new SegmentLoaderFactory(null, testUtils.getTestObjectMapper()),
         testUtils.getTestObjectMapper(),

diff --git a/...ndexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java b/...ndexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
@@ -97,6 +97,7 @@
 import org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory;
 import org.apache.druid.segment.TestHelper;
 import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.DataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
@@ -2888,6 +2889,7 @@ public void close()
         handoffNotifierFactory,
         this::makeTimeseriesOnlyConglomerate,
         Execs.directExecutor(), // queryExecutorService
+        NoopJoinableFactory.INSTANCE,
         EasyMock.createMock(MonitorScheduler.class),
         new SegmentLoaderFactory(null, testUtils.getTestObjectMapper()),
         testUtils.getTestObjectMapper(),

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolbox.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolbox.java
@@ -42,6 +42,7 @@
 import org.apache.druid.query.QueryRunnerFactoryConglomerate;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.join.JoinableFactory;
 import org.apache.druid.segment.loading.DataSegmentArchiver;
 import org.apache.druid.segment.loading.DataSegmentKiller;
 import org.apache.druid.segment.loading.DataSegmentMover;
@@ -86,6 +87,7 @@ public class TaskToolbox
   private final Provider<QueryRunnerFactoryConglomerate> queryRunnerFactoryConglomerateProvider;
   private final MonitorScheduler monitorScheduler;
   private final ExecutorService queryExecutorService;
+  private final JoinableFactory joinableFactory;
   private final SegmentLoader segmentLoader;
   private final ObjectMapper jsonMapper;
   private final File taskWorkDir;
@@ -116,6 +118,7 @@ public TaskToolbox(
       SegmentHandoffNotifierFactory handoffNotifierFactory,
       Provider<QueryRunnerFactoryConglomerate> queryRunnerFactoryConglomerateProvider,
       ExecutorService queryExecutorService,
+      JoinableFactory joinableFactory,
       MonitorScheduler monitorScheduler,
       SegmentLoader segmentLoader,
       ObjectMapper jsonMapper,
@@ -146,6 +149,7 @@ public TaskToolbox(
     this.handoffNotifierFactory = handoffNotifierFactory;
     this.queryRunnerFactoryConglomerateProvider = queryRunnerFactoryConglomerateProvider;
     this.queryExecutorService = queryExecutorService;
+    this.joinableFactory = joinableFactory;
     this.monitorScheduler = monitorScheduler;
     this.segmentLoader = segmentLoader;
     this.jsonMapper = jsonMapper;
@@ -229,6 +233,11 @@ public ExecutorService getQueryExecutorService()
     return queryExecutorService;
   }
 
+  public JoinableFactory getJoinableFactory()
+  {
+    return joinableFactory;
+  }
+
   public MonitorScheduler getMonitorScheduler()
   {
     return monitorScheduler;

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java
@@ -42,6 +42,7 @@
 import org.apache.druid.query.QueryRunnerFactoryConglomerate;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.join.JoinableFactory;
 import org.apache.druid.segment.loading.DataSegmentArchiver;
 import org.apache.druid.segment.loading.DataSegmentKiller;
 import org.apache.druid.segment.loading.DataSegmentMover;
@@ -72,6 +73,7 @@ public class TaskToolboxFactory
   private final SegmentHandoffNotifierFactory handoffNotifierFactory;
   private final Provider<QueryRunnerFactoryConglomerate> queryRunnerFactoryConglomerateProvider;
   private final ExecutorService queryExecutorService;
+  private final JoinableFactory joinableFactory;
   private final MonitorScheduler monitorScheduler;
   private final SegmentLoaderFactory segmentLoaderFactory;
   private final ObjectMapper jsonMapper;
@@ -102,6 +104,7 @@ public TaskToolboxFactory(
       SegmentHandoffNotifierFactory handoffNotifierFactory,
       Provider<QueryRunnerFactoryConglomerate> queryRunnerFactoryConglomerateProvider,
       @Processing ExecutorService queryExecutorService,
+      JoinableFactory joinableFactory,
       MonitorScheduler monitorScheduler,
       SegmentLoaderFactory segmentLoaderFactory,
       @Json ObjectMapper jsonMapper,
@@ -131,6 +134,7 @@ public TaskToolboxFactory(
     this.handoffNotifierFactory = handoffNotifierFactory;
     this.queryRunnerFactoryConglomerateProvider = queryRunnerFactoryConglomerateProvider;
     this.queryExecutorService = queryExecutorService;
+    this.joinableFactory = joinableFactory;
     this.monitorScheduler = monitorScheduler;
     this.segmentLoaderFactory = segmentLoaderFactory;
     this.jsonMapper = jsonMapper;
@@ -164,6 +168,7 @@ public TaskToolbox build(Task task)
         handoffNotifierFactory,
         queryRunnerFactoryConglomerateProvider,
         queryExecutorService,
+        joinableFactory,
         monitorScheduler,
         segmentLoaderFactory.manufacturate(taskWorkDir),
         jsonMapper,

diff --git a/.../main/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTask.java b/.../main/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTask.java
@@ -772,6 +772,7 @@ private Appenderator newAppenderator(
         toolbox.getSegmentAnnouncer(),
         toolbox.getEmitter(),
         toolbox.getQueryExecutorService(),
+        toolbox.getJoinableFactory(),
         toolbox.getCache(),
         toolbox.getCacheConfig(),
         toolbox.getCachePopulatorStats()

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/RealtimeIndexTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/RealtimeIndexTask.java
@@ -341,6 +341,7 @@ public String getVersion(final Interval interval)
         segmentPublisher,
         toolbox.getSegmentHandoffNotifierFactory(),
         toolbox.getQueryExecutorService(),
+        toolbox.getJoinableFactory(),
         toolbox.getIndexMergerV9(),
         toolbox.getIndexIO(),
         toolbox.getCache(),

diff --git a/...rvice/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java b/...rvice/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java
@@ -206,6 +206,7 @@ public Appenderator newAppenderator(FireDepartmentMetrics metrics, TaskToolbox t
         toolbox.getSegmentAnnouncer(),
         toolbox.getEmitter(),
         toolbox.getQueryExecutorService(),
+        toolbox.getJoinableFactory(),
         toolbox.getCache(),
         toolbox.getCacheConfig(),
         toolbox.getCachePopulatorStats()

diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java
@@ -34,6 +34,7 @@
 import org.apache.druid.query.QueryRunnerFactoryConglomerate;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.DataSegmentArchiver;
 import org.apache.druid.segment.loading.DataSegmentKiller;
 import org.apache.druid.segment.loading.DataSegmentMover;
@@ -109,6 +110,7 @@ public void setUp() throws IOException
         mockHandoffNotifierFactory,
         () -> mockQueryRunnerFactoryConglomerate,
         mockQueryExecutorService,
+        NoopJoinableFactory.INSTANCE,
         mockMonitorScheduler,
         mockSegmentLoaderFactory,
         ObjectMapper,

diff --git a/...t/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java b/...t/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java
@@ -113,6 +113,7 @@
 import org.apache.druid.segment.indexing.DataSchema;
 import org.apache.druid.segment.indexing.RealtimeIOConfig;
 import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.SegmentLoaderConfig;
 import org.apache.druid.segment.loading.StorageLocationConfig;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
@@ -1591,6 +1592,7 @@ public List<StorageLocationConfig> getLocations()
         handoffNotifierFactory,
         () -> conglomerate,
         Execs.directExecutor(), // queryExecutorService
+        NoopJoinableFactory.INSTANCE,
         EasyMock.createMock(MonitorScheduler.class),
         new SegmentLoaderFactory(null, testUtils.getTestObjectMapper()),
         testUtils.getTestObjectMapper(),

diff --git a/...ng-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java b/...ng-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java
@@ -61,6 +61,7 @@
 import org.apache.druid.segment.VirtualColumns;
 import org.apache.druid.segment.indexing.DataSchema;
 import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.LocalDataSegmentPuller;
 import org.apache.druid.segment.loading.LocalDataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
@@ -877,6 +878,7 @@ public List<StorageLocationConfig> getLocations()
         null,
         null,
         null,
+        NoopJoinableFactory.INSTANCE,
         null,
         loader,
         objectMapper,

diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
@@ -108,6 +108,7 @@
 import org.apache.druid.segment.indexing.RealtimeTuningConfig;
 import org.apache.druid.segment.indexing.TuningConfig;
 import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.SegmentLoadingException;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
@@ -1274,6 +1275,7 @@ private static class TestTaskToolbox extends TaskToolbox
           null,
           null,
           null,
+          NoopJoinableFactory.INSTANCE,
           null,
           null,
           null,

diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
@@ -56,6 +56,7 @@
 import org.apache.druid.metadata.TestDerbyConnector;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.LocalDataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
 import org.apache.druid.segment.loading.NoopDataSegmentKiller;
@@ -306,6 +307,7 @@ public ListenableFuture<TaskStatus> run(Task task)
             null,
             null,
             null,
+            NoopJoinableFactory.INSTANCE,
             null,
             null,
             objectMapper,

diff --git a/...ng-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java b/...ng-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java
@@ -100,6 +100,7 @@
 import org.apache.druid.segment.indexing.RealtimeIOConfig;
 import org.apache.druid.segment.indexing.RealtimeTuningConfig;
 import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.SegmentLoaderConfig;
 import org.apache.druid.segment.loading.StorageLocationConfig;
 import org.apache.druid.segment.realtime.FireDepartment;
@@ -973,6 +974,7 @@ public List<StorageLocationConfig> getLocations()
         handoffNotifierFactory,
         () -> conglomerate,
         Execs.directExecutor(), // queryExecutorService
+        NoopJoinableFactory.INSTANCE,
         EasyMock.createMock(MonitorScheduler.class),
         new SegmentLoaderFactory(null, testUtils.getTestObjectMapper()),
         testUtils.getTestObjectMapper(),

diff --git a/...service/src/test/java/org/apache/druid/indexing/common/task/TestAppenderatorsManager.java b/...service/src/test/java/org/apache/druid/indexing/common/task/TestAppenderatorsManager.java
@@ -31,6 +31,7 @@
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMerger;
 import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.join.JoinableFactory;
 import org.apache.druid.segment.loading.DataSegmentPusher;
 import org.apache.druid.segment.realtime.FireDepartmentMetrics;
 import org.apache.druid.segment.realtime.appenderator.Appenderator;
@@ -60,6 +61,7 @@ public Appenderator createRealtimeAppenderatorForTask(
       DataSegmentAnnouncer segmentAnnouncer,
       ServiceEmitter emitter,
       ExecutorService queryExecutorService,
+      JoinableFactory joinableFactory,
       Cache cache,
       CacheConfig cacheConfig,
       CachePopulatorStats cachePopulatorStats
@@ -78,6 +80,7 @@ public Appenderator createRealtimeAppenderatorForTask(
         segmentAnnouncer,
         emitter,
         queryExecutorService,
+        joinableFactory,
         cache,
         cacheConfig,
         cachePopulatorStats

diff --git a/...he/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java b/...he/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
@@ -54,6 +54,7 @@
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.metadata.EntryExistsException;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.LocalDataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
 import org.apache.druid.segment.loading.NoopDataSegmentKiller;
@@ -301,6 +302,7 @@ public File getStorageDirectory()
         null,
         null,
         null,
+        NoopJoinableFactory.INSTANCE,
         null,
         newSegmentLoader(temporaryFolder.newFolder()),
         getObjectMapper(),

diff --git a/...vice/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java b/...vice/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
@@ -33,6 +33,7 @@
 import org.apache.druid.indexing.common.task.AbstractTask;
 import org.apache.druid.indexing.common.task.NoopTask;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.NoopDataSegmentArchiver;
 import org.apache.druid.segment.loading.NoopDataSegmentKiller;
 import org.apache.druid.segment.loading.NoopDataSegmentMover;
@@ -94,6 +95,7 @@ public void setup() throws IOException
         null,
         null,
         null,
+        NoopJoinableFactory.INSTANCE,
         null,
         new SegmentLoaderFactory(null, utils.getTestObjectMapper()),
         utils.getTestObjectMapper(),

diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
@@ -117,6 +117,7 @@
 import org.apache.druid.segment.indexing.RealtimeIOConfig;
 import org.apache.druid.segment.indexing.RealtimeTuningConfig;
 import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.DataSegmentArchiver;
 import org.apache.druid.segment.loading.DataSegmentMover;
 import org.apache.druid.segment.loading.DataSegmentPusher;
@@ -663,6 +664,7 @@ public void unannounceSegments(Iterable<DataSegment> segments)
         handoffNotifierFactory,
         () -> queryRunnerFactoryConglomerate, // query runner factory conglomerate corporation unionized collective
         Execs.directExecutor(), // query executor service
+        NoopJoinableFactory.INSTANCE,
         monitorScheduler, // monitor scheduler
         new SegmentLoaderFactory(null, new DefaultObjectMapper()),
         MAPPER,
@@ -1329,6 +1331,7 @@ public void testUnifiedAppenderatorsManagerCleanup() throws Exception
 
     UnifiedIndexerAppenderatorsManager unifiedIndexerAppenderatorsManager = new UnifiedIndexerAppenderatorsManager(
         exec,
+        NoopJoinableFactory.INSTANCE,
         new WorkerConfig(),
         MapCache.create(2048),
         new CacheConfig(),

diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskManagerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskManagerTest.java
@@ -40,6 +40,7 @@
 import org.apache.druid.java.util.common.FileUtils;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.loading.SegmentLoaderConfig;
 import org.apache.druid.segment.loading.StorageLocationConfig;
 import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
@@ -120,6 +121,7 @@ public List<StorageLocationConfig> getLocations()
                 notifierFactory,
                 null,
                 null,
+                NoopJoinableFactory.INSTANCE,
                 null,
                 new SegmentLoaderFactory(null, jsonMapper),
                 jsonMapper,

diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskMonitorTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskMonitorTest.java
@@ -47,6 +47,7 @@
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.join.NoopJoinableFactory;
 import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.initialization.IndexerZkConfig;
@@ -170,7 +171,18 @@ private WorkerTaskMonitor createTaskMonitor()
                 taskConfig,
                 null,
                 taskActionClientFactory,
-                null, null, null, null, null, null, null, notifierFactory, null, null, null,
+                null,
+                null,
+                null,
+                null,
+                null,
+                null,
+                null,
+                notifierFactory,
+                null,
+                null,
+                NoopJoinableFactory.INSTANCE,
+                null,
                 new SegmentLoaderFactory(null, jsonMapper),
                 jsonMapper,
                 indexIO,