You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "gianm (via GitHub)" <gi...@apache.org> on 2024/03/19 09:49:28 UTC
[PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
gianm opened a new pull request, #16168:
URL: https://github.com/apache/druid/pull/16168
This patch contains two controller changes that make progress towards a lower-latency MSQ.
These are both larger changes, but I developed them together and it was most straightforward to put them into a single PR rather than separating into multiple PRs, especially given they both involved changes in common files like ControllerImpl and ControllerContext.
**Key classes:**
These files are the most interesting IMO.
- Controller
- ControllerImpl
- ControllerQueryKernel
- ControllerQueryKernelConfig
- ControllerUtils (especially computeStageGroups)
- ControllerContext
- MSQWorkerTaskLauncher
**Key changes:**
First, support for **in-memory shuffles**. The main feature of in-memory shuffles, as far as the controller is concerned, is that they are not fully buffered. That means that whenever a producer stage uses in-memory output, its consumer must run concurrently. The controller determines which stages run concurrently, and when they start and stop.
"Leapfrogging" allows any chain of sort-based stages to use in-memory shuffles even if we can only run two stages at once. For example, in a linear chain of stages 0 -> 1 -> 2 where all do sort-based shuffles, we can use in-memory shuffling for each one while only running two at once. (When stage 1 is done reading input and about to start writing its output, we can stop 0 and start 2.)
1) New `OutputChannelMode` enum attached to `WorkOrder` that tells workers
whether stage output should be in memory (`MEMORY`), or use local or durable
storage.
2) New logic in the `ControllerQueryKernel` to determine which stages can use
in-memory shuffling (`ControllerUtils#computeStageGroups`) and to launch them
at the appropriate time (`ControllerQueryKernel#createNewKernels`).
3) New `doneReadingInput` method on `Controller` (passed down to the stage kernels)
which allows stages to transition to `POST_READING` even if they are not
gathering statistics. This is important because it enables "leapfrogging"
for `HASH_LOCAL_SORT` shuffles, and for `GLOBAL_SORT` shuffles with 1 partition.
4) Moved result-reading from `ControllerContext#writeReports` to new `QueryListener`
interface, which `ControllerImpl` feeds results to row-by-row while the query
is still running. Important so we can read query results from the final
stage using an in-memory channel.
5) New class `ControllerQueryKernelConfig` holds configs that control kernel
behavior (such as whether to pipeline, maximum number of concurrent stages,
etc). Generated by the `ControllerContext`.
Second, a refactor **towards running workers in persistent JVMs** that are able to cache data across queries. This is helpful because I believe we'll want to reuse JVMs and cached data for latency reasons.
1) Move creation of `WorkerManager` and `TableInputSpecSlicer` to the
`ControllerContext`, rather than `ControllerImpl`. This allows managing workers and
work assignment differently when JVMs are reusable.
2) Lift the Controller Jersey resource out from `ControllerChatHandler` to a
reusable resource `ControllerResource`.
3) Move memory introspection to a `MemoryIntrospector` interface, and introduce
`ControllerMemoryParameters` that uses it. This makes it easier to run MSQ in
process types other than Indexer and Peon.
Both of these areas will have follow-ups that make similar changes on the worker side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1579245838
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerContext.java:
##########
@@ -21,24 +21,44 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.inject.Injector;
-import org.apache.druid.client.coordinator.CoordinatorClient;
-import org.apache.druid.indexer.report.TaskReport;
import org.apache.druid.indexing.common.actions.TaskActionClient;
import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.msq.indexing.MSQSpec;
+import org.apache.druid.msq.input.InputSpecSlicer;
+import org.apache.druid.msq.input.table.SegmentsInputSlice;
+import org.apache.druid.msq.input.table.TableInputSpec;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.controller.ControllerQueryKernelConfig;
import org.apache.druid.server.DruidNode;
/**
- * Context used by multi-stage query controllers.
- *
- * Useful because it allows test fixtures to provide their own implementations.
+ * Context used by multi-stage query controllers. Useful because it allows test fixtures to provide their own
+ * implementations.
*/
public interface ControllerContext
{
- ServiceEmitter emitter();
+ /**
+ * Configuration for {@link org.apache.druid.msq.kernel.controller.ControllerQueryKernel}.
+ */
+ ControllerQueryKernelConfig queryKernelConfig(MSQSpec querySpec, QueryDefinition queryDef);
+ /**
+ * Callback from the controller implementation to "register" the controller. Used in the indexing task implementation
+ * to set up the task chat web service.
+ */
+ void registerController(Controller controller, Closer closer);
Review Comment:
The controller context should be owned by a controller. This method feels weird to me . If we need to register a closer, then we can add a addCloser() method to this interface no ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1575633493
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
Review Comment:
That makes sense. I have changed it.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/StageGroup.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.kernel.StageId;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Group of stages that must be launched as a unit. Within each group, stages communicate with each other using
+ * {@link OutputChannelMode#MEMORY} channels. The final stage in a group writes its own output using
+ * {@link #lastStageOutputChannelMode()}.
+ */
+public class StageGroup
+{
+ private final List<StageId> stageIds;
+ private final OutputChannelMode groupOutputChannelMode;
+
+ public StageGroup(final List<StageId> stageIds, final OutputChannelMode groupOutputChannelMode)
+ {
+ this.stageIds = stageIds;
+ this.groupOutputChannelMode = groupOutputChannelMode;
+ }
+
+ /**
+ * List of stage IDs in this group.
Review Comment:
That's true. I have added a comment.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/StageGroup.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.kernel.StageId;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Group of stages that must be launched as a unit. Within each group, stages communicate with each other using
+ * {@link OutputChannelMode#MEMORY} channels. The final stage in a group writes its own output using
+ * {@link #lastStageOutputChannelMode()}.
+ */
+public class StageGroup
+{
+ private final List<StageId> stageIds;
Review Comment:
Yes, it is required to be linear at this point (mostly for simplicity). It could be changed in the future. I have added a comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1581235367
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MemoryIntrospector.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.exec;
+
+import org.apache.druid.msq.kernel.WorkOrder;
+
+/**
+ * Introspector used to generate {@link ControllerMemoryParameters}.
+ */
+public interface MemoryIntrospector
+{
+ /**
+ * Amount of total memory in the entire JVM.
+ */
+ long totalMemoryInJvm();
+
+ /**
+ * Amount of memory usable for the multi-stage query engine in the entire JVM.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long usableMemoryInJvm();
+
+ /**
+ * Amount of total JVM memory required for a particular amount of usable memory to be available.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long jvmMemoryRequiredForUsableMemory(long usableMemory);
Review Comment:
Sounds good, I changed the name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1575633381
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
Review Comment:
I don't think so. I added a comment with a reason why:
```
// Note: priorGroup.size() is strictly less than config.getMaxConcurrentStages(), because the prior group
// would have its size limited by maxStageGroupSizeAllowingForDownstreamConsumer below.
```
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
Review Comment:
The final stage is the only one that can possibly sort, because of the check below that closes off the stage group if currentStageId "doesSortDuringShuffle()". I think the logic here would apply if _any_ stage in the group sorted, though. (If any stage sorts, that causes a pipeline break that means the prior stage group would no longer be needed.)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
Review Comment:
I added a few paragraphs of class-level javadoc to discuss how "leapfrogging" works- that is the scenario where this can happen. I also added a mention here:
```
// Prior group must run concurrently with this group. (Can happen when leapfrogging; see class-level Javadoc.)
```
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
Review Comment:
Hmm, as far as I know, this logic doesn't assume the keySet is sorted. It's nice if it is, since it means the execution happens in a consistent order each time this method is run, but I don't think it's required for correctness. Unless I'm missing something.
At any rate, we _do_ use sorted sets, because it _is_ nice for the order to be consistent. There is no harm in tightening up the contract, so I did, to `SortedSet`.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
Review Comment:
Sounds good. I made this change.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
Review Comment:
It means the stage input data is fully available, which means it would be OK to start the stage. I have added a comment about this:
```
inflow.get(stageId).isEmpty() /* stage data is fully available */
```
And also another comment above the whole block describing what this code is doing:
```
// This loop simulates execution of all stages, such that we arrive at an order of execution that is compatible
// with all relevant constraints.
```
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
+ } else {
+ // When the current group doesn't sort, we can't leapfrog.
+ maxStageGroupSizeAllowingForDownstreamConsumer = maxStageGroupSize - 1;
+ }
+
+ currentOutputChannelMode =
+ getOutputChannelMode(
+ queryDef,
+ currentStageId.getStageNumber(),
+ config.getDestination().toSelectDestination(),
+ config.isDurableStorage(),
+ canStream
+
+ // Stages can only stream if they have <= 1 downstream stage (checked by "canStreamOutput") and if
+ // that downstream stage has all of its other inputs available.
+ && (currentOutflow.isEmpty() ||
+ Collections.singleton(currentStageId)
+ .equals(inflow.get(Iterables.getOnlyElement(currentOutflow))))
+
+ // Stages can only stream if that one downstream consumer is able to start concurrently. So, once the
+ // stage group gets too big, we must stop streaming.
+ && (currentOutflow.isEmpty()
+ || currentStageGroup.size() < maxStageGroupSizeAllowingForDownstreamConsumer)
+ );
+
+ currentStageGroup.add(currentStageId);
+
+ if (currentOutflow.size() == 1
+ && currentStageGroup.size() < maxStageGroupSize
+ && currentOutputChannelMode == OutputChannelMode.MEMORY
+
+ // Sorting causes a pipeline break: a sorting stage must read all its input before writing any output.
+ // Continue the stage group only if this stage does not sort its output.
+ && !queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ currentStageId = Iterables.getOnlyElement(currentOutflow);
+ } else {
+ currentStageId = null;
+ }
+ }
+
+ stageGroups.add(new StageGroup(currentStageGroup, currentOutputChannelMode));
+
+ for (final StageId stageId : currentStageGroup) {
+ stagesRun.add(stageId);
+ removeStageFlow(stageId, inflow, outflow);
+ }
+ }
+ }
+
+ return stageGroups;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that flow *into* that stage. Uses TreeMaps and TreeSets so we have a
+ * consistent order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageInflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>()).add(inputStageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that depend on that stage. Uses TreeMaps and TreeSets so we have a consistent
+ * order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(inputStageId, ignored -> new TreeSet<>()).add(stageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Whether output of a stage can possibly use {@link OutputChannelMode#MEMORY}. Returning true does not guarantee
+ * that the stage *will* use {@link OutputChannelMode#MEMORY}. Additional requirements are checked in
+ * {@link #computeStageGroups(QueryDefinition, ControllerQueryKernelConfig)}.
+ */
+ public static boolean canStreamOutput(
+ final QueryDefinition queryDefinition,
+ final int stageNumber,
+ final ControllerQueryKernelConfig config,
+ final Map<StageId, Set<StageId>> outflowMap
+ )
+ {
+ if (config.isFaultTolerant()) {
+ // Cannot stream if fault tolerance is enabled: durable storage is required.
+ return false;
+ }
+
+ if (!config.isPipeline() || config.getMaxConcurrentStages() < 2) {
+ // Cannot stream if pipelining (& running multiple stages at once) is not enabled.
+ return false;
+ }
+
+ final StageId stageId = queryDefinition.getStageDefinition(stageNumber).getId();
+ final Set<StageId> outflowStageIds = outflowMap.get(stageId);
+
+ if (outflowStageIds.isEmpty()) {
+ return true;
+ } else if (outflowStageIds.size() == 1) {
+ final StageDefinition outflowStageDef =
+ queryDefinition.getStageDefinition(Iterables.getOnlyElement(outflowStageIds));
+
+ // Two things happening here:
+ // 1) Stages cannot both stream and broadcast their output. This is because when streaming, we can only
+ // support a single reader.
+ // 2) Stages can only receive a single streamed input. This isn't strictly necessary, but it simplifies the
Review Comment:
Ah, I really meant this to refer to output channel mode MEMORY. I have tried to make it more clear and have removed the word "stream". This comment now reads:
```
// Two things happening here:
// 1) Stages cannot use output mode MEMORY when broadcasting. This is because when using output mode MEMORY, we
// can only support a single reader.
// 2) Downstream stages can only have a single input stage with output mode MEMORY. This isn't strictly
// necessary, but it simplifies the logic around concurrently launching stages.
```
I have also renamed various variables, like `canStream` to `canUseMemoryOutput`.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
+ } else {
+ // When the current group doesn't sort, we can't leapfrog.
+ maxStageGroupSizeAllowingForDownstreamConsumer = maxStageGroupSize - 1;
+ }
+
+ currentOutputChannelMode =
+ getOutputChannelMode(
+ queryDef,
+ currentStageId.getStageNumber(),
+ config.getDestination().toSelectDestination(),
+ config.isDurableStorage(),
+ canStream
+
+ // Stages can only stream if they have <= 1 downstream stage (checked by "canStreamOutput") and if
+ // that downstream stage has all of its other inputs available.
+ && (currentOutflow.isEmpty() ||
+ Collections.singleton(currentStageId)
+ .equals(inflow.get(Iterables.getOnlyElement(currentOutflow))))
+
+ // Stages can only stream if that one downstream consumer is able to start concurrently. So, once the
+ // stage group gets too big, we must stop streaming.
+ && (currentOutflow.isEmpty()
+ || currentStageGroup.size() < maxStageGroupSizeAllowingForDownstreamConsumer)
+ );
+
+ currentStageGroup.add(currentStageId);
+
+ if (currentOutflow.size() == 1
+ && currentStageGroup.size() < maxStageGroupSize
+ && currentOutputChannelMode == OutputChannelMode.MEMORY
+
+ // Sorting causes a pipeline break: a sorting stage must read all its input before writing any output.
+ // Continue the stage group only if this stage does not sort its output.
+ && !queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ currentStageId = Iterables.getOnlyElement(currentOutflow);
+ } else {
+ currentStageId = null;
+ }
+ }
+
+ stageGroups.add(new StageGroup(currentStageGroup, currentOutputChannelMode));
+
+ for (final StageId stageId : currentStageGroup) {
+ stagesRun.add(stageId);
+ removeStageFlow(stageId, inflow, outflow);
+ }
+ }
+ }
+
+ return stageGroups;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that flow *into* that stage. Uses TreeMaps and TreeSets so we have a
+ * consistent order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageInflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>()).add(inputStageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that depend on that stage. Uses TreeMaps and TreeSets so we have a consistent
+ * order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(inputStageId, ignored -> new TreeSet<>()).add(stageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Whether output of a stage can possibly use {@link OutputChannelMode#MEMORY}. Returning true does not guarantee
+ * that the stage *will* use {@link OutputChannelMode#MEMORY}. Additional requirements are checked in
+ * {@link #computeStageGroups(QueryDefinition, ControllerQueryKernelConfig)}.
+ */
+ public static boolean canStreamOutput(
+ final QueryDefinition queryDefinition,
+ final int stageNumber,
+ final ControllerQueryKernelConfig config,
+ final Map<StageId, Set<StageId>> outflowMap
+ )
+ {
+ if (config.isFaultTolerant()) {
+ // Cannot stream if fault tolerance is enabled: durable storage is required.
+ return false;
+ }
+
+ if (!config.isPipeline() || config.getMaxConcurrentStages() < 2) {
+ // Cannot stream if pipelining (& running multiple stages at once) is not enabled.
+ return false;
+ }
+
+ final StageId stageId = queryDefinition.getStageDefinition(stageNumber).getId();
+ final Set<StageId> outflowStageIds = outflowMap.get(stageId);
+
+ if (outflowStageIds.isEmpty()) {
+ return true;
+ } else if (outflowStageIds.size() == 1) {
+ final StageDefinition outflowStageDef =
+ queryDefinition.getStageDefinition(Iterables.getOnlyElement(outflowStageIds));
+
+ // Two things happening here:
+ // 1) Stages cannot both stream and broadcast their output. This is because when streaming, we can only
+ // support a single reader.
+ // 2) Stages can only receive a single streamed input. This isn't strictly necessary, but it simplifies the
+ // logic around concurrently launching stages.
+ return stageId.equals(getOnlyNonBroadcastInputAsStageId(outflowStageDef));
+ } else {
+ return false;
+ }
+ }
+
+ /**
+ * Return an {@link OutputChannelMode} to use for a given stage, based on query and context.
+ */
+ public static OutputChannelMode getOutputChannelMode(
+ final QueryDefinition queryDef,
+ final int stageNumber,
+ @Nullable final MSQSelectDestination selectDestination,
+ final boolean durableStorage,
+ final boolean canStream
+ )
+ {
+ final boolean isFinalStage = queryDef.getFinalStageDefinition().getStageNumber() == stageNumber;
+
+ if (isFinalStage && selectDestination == MSQSelectDestination.DURABLESTORAGE) {
+ return OutputChannelMode.DURABLE_STORAGE_QUERY_RESULTS;
+ } else if (canStream) {
+ return OutputChannelMode.MEMORY;
+ } else if (durableStorage) {
+ return OutputChannelMode.DURABLE_STORAGE_INTERMEDIATE;
+ } else {
+ return OutputChannelMode.LOCAL_STORAGE;
+ }
+ }
+
+ /**
+ * If a stage has a single non-broadcast input stage, returns that input stage. Otherwise, returns null.
+ * This is a helper used by {@link #canStreamOutput}.
+ */
+ @Nullable
+ public static StageId getOnlyNonBroadcastInputAsStageId(final StageDefinition downstreamStageDef)
+ {
+ final List<InputSpec> inputSpecs = downstreamStageDef.getInputSpecs();
+ final IntSet broadcastInputNumbers = downstreamStageDef.getBroadcastInputNumbers();
+
+ if (inputSpecs.size() - broadcastInputNumbers.size() != 1) {
+ return null;
+ }
+
+ for (int i = 0; i < inputSpecs.size(); i++) {
+ if (!broadcastInputNumbers.contains(i)) {
+ final IntSet stageNumbers = InputSpecs.getStageNumbers(Collections.singletonList(inputSpecs.get(i)));
+ if (stageNumbers.size() == 1) {
+ return new StageId(downstreamStageDef.getId().getQueryId(), stageNumbers.iterator().nextInt());
+ }
+ }
+ }
+
+ return null;
+ }
+
+ private static void removeStageFlow(
Review Comment:
I have added javadocs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1583394775
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MemoryIntrospector.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.exec;
+
+import org.apache.druid.msq.kernel.WorkOrder;
+
+/**
+ * Introspector used to generate {@link ControllerMemoryParameters}.
+ */
+public interface MemoryIntrospector
+{
+ /**
+ * Amount of total memory in the entire JVM.
+ */
+ long totalMemoryInJvm();
+
+ /**
+ * Amount of memory usable for the multi-stage query engine in the entire JVM.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long usableMemoryInJvm();
+
+ /**
+ * Amount of total JVM memory required for a particular amount of usable memory to be available.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long jvmMemoryRequiredForUsableMemory(long usableMemory);
+
+ /**
+ * Maximum number of queries that run simultaneously in this JVM.
+ *
+ * On workers, this is the maximum number of {@link Worker} that run simultaneously. See
+ * {@link WorkerMemoryParameters} for how memory is divided among and within {@link WorkOrder} run by a worker.
Review Comment:
It's really meant to mean "within one JVM". So in a task it's always 1. I will update it to be clearer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1530710796
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -2995,15 +2873,14 @@
*/
private interface TaskContactFn
{
- ListenableFuture<Void> contactTask(WorkerClient client, String taskId, int workerNumber);
+ ListenableFuture<Void> contactTask(WorkerClient client, String workerId, int workerNumber);
}
/**
* Interface used when {@link TaskContactFn#contactTask(WorkerClient, String, int)} returns a successful future.
*/
private interface TaskContactSuccess
{
- void onSuccess(String taskId);
-
+ void onSuccess(String workerId, int workerNumber);
Review Comment:
## Useless parameter
The parameter 'workerId' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/7150)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm merged PR #16168:
URL: https://github.com/apache/druid/pull/16168
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1558115291
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/report/MSQResultsReport.java:
##########
@@ -132,9 +72,9 @@
}
@JsonProperty("results")
- public Yielder<Object[]> getResultYielder()
+ public List<Object[]> getResults()
Review Comment:
We trust the caller to not modify the arrays here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1575632292
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
Review Comment:
I extended the comment to say:
```
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
Review Comment:
I extended the comment to say:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1558117581
##########
extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/kernel/controller/MockQueryDefinitionBuilder.java:
##########
@@ -113,27 +126,60 @@
);
definedStages.add(stageNumber);
- ShuffleSpec shuffleSpec;
+ ShuffleSpec shuffleSpec = null;
- if (shuffling) {
- shuffleSpec = new GlobalSortMaxCountShuffleSpec(
- new ClusterBy(
- ImmutableList.of(
- new KeyColumn(SHUFFLE_KEY_COLUMN, KeyOrder.ASCENDING)
+ if (shuffleKind != null) {
+ switch (shuffleKind) {
+ case GLOBAL_SORT:
+ shuffleSpec = new GlobalSortMaxCountShuffleSpec(
+ new ClusterBy(
+ ImmutableList.of(
+ new KeyColumn(SHUFFLE_KEY_COLUMN, KeyOrder.ASCENDING)
+ ),
+ 0
),
- 0
- ),
- MAX_NUM_PARTITIONS,
- false
- );
- } else {
- shuffleSpec = null;
+ MAX_NUM_PARTITIONS,
+ false
+ );
+ break;
+
+ case HASH_LOCAL_SORT:
+ case HASH:
+ shuffleSpec = new HashShuffleSpec(
+ new ClusterBy(
+ ImmutableList.of(
+ new KeyColumn(
+ SHUFFLE_KEY_COLUMN,
+ shuffleKind == ShuffleKind.HASH ? KeyOrder.NONE : KeyOrder.ASCENDING
+ )
+ ),
+ 0
+ ),
+ MAX_NUM_PARTITIONS
+ );
+ break;
+
+ case MIX:
+ shuffleSpec = MixShuffleSpec.instance();
+ break;
+ }
+
+ if (shuffleKind != shuffleSpec.kind()) {
Review Comment:
fixed.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerContext.java:
##########
@@ -53,32 +71,33 @@
DruidNode selfNode();
/**
- * Provide access to the Coordinator service.
+ * Provides an {@link InputSpecSlicer} that slices {@link TableInputSpec} into {@link SegmentsInputSlice}.
*/
- CoordinatorClient coordinatorClient();
+ InputSpecSlicer newTableInputSpecSlicer(WorkerManager workerManager);
Review Comment:
removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1583396531
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/guice/PeonMemoryManagementModule.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.guice;
+
+import com.google.inject.Binder;
+import com.google.inject.Provides;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.frame.processor.Bouncer;
+import org.apache.druid.guice.LazySingleton;
+import org.apache.druid.guice.annotations.LoadScope;
+import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.msq.exec.MemoryIntrospector;
+import org.apache.druid.msq.exec.MemoryIntrospectorImpl;
+import org.apache.druid.query.lookup.LookupExtractorFactoryContainerProvider;
+import org.apache.druid.utils.JvmUtils;
+
+/**
+ * Provides {@link MemoryIntrospector} for single-task-per-JVM model.
+ *
+ * @see IndexerMemoryManagementModule for multi-task-per-JVM model used on {@link org.apache.druid.cli.CliIndexer}
+ */
+@LoadScope(roles = NodeRole.PEON_JSON_NAME)
+public class PeonMemoryManagementModule implements DruidModule
+{
+ /**
+ * Peons have a single worker per JVM.
+ */
+ private static final int NUM_WORKERS_IN_JVM = 1;
+
+ /**
+ * Peons may have more than one processing thread, but we currently only use one of them.
+ */
+ private static final int NUM_PROCESSING_THREADS = 1;
Review Comment:
For the Peons, I think we can start using more threads. I am planning on changing that in a future patch, probably the follow-up to this patch that does the Worker changes.
For the Indexer, I'm not sure the current resource model actually makes sense. It probably needs some adjustment. For now, I am just keeping it the way it is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1583397249
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/guice/IndexerMemoryManagementModule.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.guice;
+
+import com.google.inject.Binder;
+import com.google.inject.Provides;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.frame.processor.Bouncer;
+import org.apache.druid.guice.LazySingleton;
+import org.apache.druid.guice.annotations.LoadScope;
+import org.apache.druid.indexing.worker.config.WorkerConfig;
+import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.msq.exec.MemoryIntrospector;
+import org.apache.druid.msq.exec.MemoryIntrospectorImpl;
+import org.apache.druid.query.DruidProcessingConfig;
+import org.apache.druid.query.lookup.LookupExtractorFactoryContainerProvider;
+import org.apache.druid.utils.JvmUtils;
+
+/**
+ * Provides {@link MemoryIntrospector} for multi-task-per-JVM model.
+ *
+ * @see PeonMemoryManagementModule for single-task-per-JVM model used on {@link org.apache.druid.cli.CliPeon}
+ */
+@LoadScope(roles = NodeRole.INDEXER_JSON_NAME)
+public class IndexerMemoryManagementModule implements DruidModule
+{
+ /**
+ * Allocate up to 75% of memory for MSQ-related stuff (if all running tasks are MSQ tasks).
+ */
+ private static final double USABLE_MEMORY_FRACTION = 0.75;
+
+ @Override
+ public void configure(Binder binder)
+ {
+ // Nothing to do.
+ }
+
+ @Provides
+ @LazySingleton
Review Comment:
Sounds reasonable. I changed the name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1575633139
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
Review Comment:
It's prevented by the fact that when there's a pipeline break (i.e. some stage that sorts) in a stage group, the upstream stage group is closed before the downstream stage group is started. So the `priorGroup.size()` and the `1` will not be happening simultaneously.
I extended the comment to say:
```
// When the current group sorts, there's a pipeline break, so we can leapfrog: close the prior group before
// starting the downstream group. In this case, we only need to reserve a single concurrent-stage slot for
// a downstream consumer.
```
The new class-level javadoc should help shine light on this too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1579245838
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerContext.java:
##########
@@ -21,24 +21,44 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.inject.Injector;
-import org.apache.druid.client.coordinator.CoordinatorClient;
-import org.apache.druid.indexer.report.TaskReport;
import org.apache.druid.indexing.common.actions.TaskActionClient;
import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.msq.indexing.MSQSpec;
+import org.apache.druid.msq.input.InputSpecSlicer;
+import org.apache.druid.msq.input.table.SegmentsInputSlice;
+import org.apache.druid.msq.input.table.TableInputSpec;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.controller.ControllerQueryKernelConfig;
import org.apache.druid.server.DruidNode;
/**
- * Context used by multi-stage query controllers.
- *
- * Useful because it allows test fixtures to provide their own implementations.
+ * Context used by multi-stage query controllers. Useful because it allows test fixtures to provide their own
+ * implementations.
*/
public interface ControllerContext
{
- ServiceEmitter emitter();
+ /**
+ * Configuration for {@link org.apache.druid.msq.kernel.controller.ControllerQueryKernel}.
+ */
+ ControllerQueryKernelConfig queryKernelConfig(MSQSpec querySpec, QueryDefinition queryDef);
+ /**
+ * Callback from the controller implementation to "register" the controller. Used in the indexing task implementation
+ * to set up the task chat web service.
+ */
+ void registerController(Controller controller, Closer closer);
Review Comment:
The controller context should be owned by a controller. This method feels weird to me . If we need to register a closer, then we can add a addCloser() method to this interface no ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1558114216
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/rpc/ControllerResource.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.rpc;
+
+import org.apache.druid.indexing.common.TaskReport;
+import org.apache.druid.msq.counters.CounterSnapshots;
+import org.apache.druid.msq.counters.CounterSnapshotsTree;
+import org.apache.druid.msq.exec.Controller;
+import org.apache.druid.msq.exec.ControllerClient;
+import org.apache.druid.msq.indexing.MSQTaskList;
+import org.apache.druid.msq.indexing.error.MSQErrorReport;
+import org.apache.druid.msq.kernel.StageId;
+import org.apache.druid.msq.statistics.PartialKeyStatisticsInformation;
+import org.apache.druid.server.security.AuthorizerMapper;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.util.List;
+import java.util.Map;
+
+public class ControllerResource
+{
+ private final Controller controller;
+ private final ResourcePermissionMapper permissionMapper;
+ private final AuthorizerMapper authorizerMapper;
+
+ public ControllerResource(
+ final Controller controller,
+ final ResourcePermissionMapper permissionMapper,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.controller = controller;
+ this.permissionMapper = permissionMapper;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ /**
+ * Used by subtasks to post {@link PartialKeyStatisticsInformation} for shuffling stages.
+ *
+ * See {@link ControllerClient#postPartialKeyStatistics(StageId, int, PartialKeyStatisticsInformation)}
+ * for the client-side code that calls this API.
+ */
+ @POST
+ @Path("/partialKeyStatisticsInformation/{queryId}/{stageNumber}/{workerNumber}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response httpPostPartialKeyStatistics(
+ final Object partialKeyStatisticsObject,
+ @PathParam("queryId") final String queryId,
+ @PathParam("stageNumber") final int stageNumber,
+ @PathParam("workerNumber") final int workerNumber,
+ @Context final HttpServletRequest req
+ )
+ {
+ MSQResourceUtils.authorizeAdminRequest(permissionMapper, authorizerMapper, req);
+ controller.updatePartialKeyStatisticsInformation(stageNumber, workerNumber, partialKeyStatisticsObject);
+ return Response.status(Response.Status.ACCEPTED).build();
+ }
+
+ /**
+ * Used by subtasks to post system errors. Note that the errors are organized by taskId, not by query/stage/worker,
+ * because system errors are associated with a task rather than a specific query/stage/worker execution context.
+ *
+ * See {@link ControllerClient#postWorkerError} for the client-side code that calls this API.
+ */
+ @POST
+ @Path("/workerError/{taskId}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response httpPostWorkerError(
+ final MSQErrorReport errorReport,
+ @PathParam("taskId") final String taskId,
Review Comment:
This code was just moved; the parameter is unused in master as well
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/rpc/ControllerResource.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.rpc;
+
+import org.apache.druid.indexing.common.TaskReport;
+import org.apache.druid.msq.counters.CounterSnapshots;
+import org.apache.druid.msq.counters.CounterSnapshotsTree;
+import org.apache.druid.msq.exec.Controller;
+import org.apache.druid.msq.exec.ControllerClient;
+import org.apache.druid.msq.indexing.MSQTaskList;
+import org.apache.druid.msq.indexing.error.MSQErrorReport;
+import org.apache.druid.msq.kernel.StageId;
+import org.apache.druid.msq.statistics.PartialKeyStatisticsInformation;
+import org.apache.druid.server.security.AuthorizerMapper;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.util.List;
+import java.util.Map;
+
+public class ControllerResource
+{
+ private final Controller controller;
+ private final ResourcePermissionMapper permissionMapper;
+ private final AuthorizerMapper authorizerMapper;
+
+ public ControllerResource(
+ final Controller controller,
+ final ResourcePermissionMapper permissionMapper,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.controller = controller;
+ this.permissionMapper = permissionMapper;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ /**
+ * Used by subtasks to post {@link PartialKeyStatisticsInformation} for shuffling stages.
+ *
+ * See {@link ControllerClient#postPartialKeyStatistics(StageId, int, PartialKeyStatisticsInformation)}
+ * for the client-side code that calls this API.
+ */
+ @POST
+ @Path("/partialKeyStatisticsInformation/{queryId}/{stageNumber}/{workerNumber}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response httpPostPartialKeyStatistics(
+ final Object partialKeyStatisticsObject,
+ @PathParam("queryId") final String queryId,
Review Comment:
This code was just moved; the parameter is unused in master as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1576196553
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MemoryIntrospector.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.exec;
+
+import org.apache.druid.msq.kernel.WorkOrder;
+
+/**
+ * Introspector used to generate {@link ControllerMemoryParameters}.
+ */
+public interface MemoryIntrospector
+{
+ /**
+ * Amount of total memory in the entire JVM.
+ */
+ long totalMemoryInJvm();
+
+ /**
+ * Amount of memory usable for the multi-stage query engine in the entire JVM.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long usableMemoryInJvm();
+
+ /**
+ * Amount of total JVM memory required for a particular amount of usable memory to be available.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long jvmMemoryRequiredForUsableMemory(long usableMemory);
Review Comment:
```suggestion
long computeJVMMemoryRequiredForUsableMemory(long usableMemory);
```
the suggest name makes it clearer about the potential cost of this operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1530125108
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerContext.java:
##########
@@ -53,32 +71,33 @@
DruidNode selfNode();
/**
- * Provide access to the Coordinator service.
+ * Provides an {@link InputSpecSlicer} that slices {@link TableInputSpec} into {@link SegmentsInputSlice}.
*/
- CoordinatorClient coordinatorClient();
+ InputSpecSlicer newTableInputSpecSlicer(WorkerManager workerManager);
Review Comment:
## Useless parameter
The parameter 'workerManager' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/7149)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/rpc/ControllerResource.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.rpc;
+
+import org.apache.druid.indexing.common.TaskReport;
+import org.apache.druid.msq.counters.CounterSnapshots;
+import org.apache.druid.msq.counters.CounterSnapshotsTree;
+import org.apache.druid.msq.exec.Controller;
+import org.apache.druid.msq.exec.ControllerClient;
+import org.apache.druid.msq.indexing.MSQTaskList;
+import org.apache.druid.msq.indexing.error.MSQErrorReport;
+import org.apache.druid.msq.kernel.StageId;
+import org.apache.druid.msq.statistics.PartialKeyStatisticsInformation;
+import org.apache.druid.server.security.AuthorizerMapper;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.util.List;
+import java.util.Map;
+
+public class ControllerResource
+{
+ private final Controller controller;
+ private final ResourcePermissionMapper permissionMapper;
+ private final AuthorizerMapper authorizerMapper;
+
+ public ControllerResource(
+ final Controller controller,
+ final ResourcePermissionMapper permissionMapper,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.controller = controller;
+ this.permissionMapper = permissionMapper;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ /**
+ * Used by subtasks to post {@link PartialKeyStatisticsInformation} for shuffling stages.
+ *
+ * See {@link ControllerClient#postPartialKeyStatistics(StageId, int, PartialKeyStatisticsInformation)}
+ * for the client-side code that calls this API.
+ */
+ @POST
+ @Path("/partialKeyStatisticsInformation/{queryId}/{stageNumber}/{workerNumber}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response httpPostPartialKeyStatistics(
+ final Object partialKeyStatisticsObject,
+ @PathParam("queryId") final String queryId,
+ @PathParam("stageNumber") final int stageNumber,
+ @PathParam("workerNumber") final int workerNumber,
+ @Context final HttpServletRequest req
+ )
+ {
+ MSQResourceUtils.authorizeAdminRequest(permissionMapper, authorizerMapper, req);
+ controller.updatePartialKeyStatisticsInformation(stageNumber, workerNumber, partialKeyStatisticsObject);
+ return Response.status(Response.Status.ACCEPTED).build();
+ }
+
+ /**
+ * Used by subtasks to post system errors. Note that the errors are organized by taskId, not by query/stage/worker,
+ * because system errors are associated with a task rather than a specific query/stage/worker execution context.
+ *
+ * See {@link ControllerClient#postWorkerError} for the client-side code that calls this API.
+ */
+ @POST
+ @Path("/workerError/{taskId}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response httpPostWorkerError(
+ final MSQErrorReport errorReport,
+ @PathParam("taskId") final String taskId,
Review Comment:
## Useless parameter
The parameter 'taskId' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/7151)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/rpc/ControllerResource.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.rpc;
+
+import org.apache.druid.indexing.common.TaskReport;
+import org.apache.druid.msq.counters.CounterSnapshots;
+import org.apache.druid.msq.counters.CounterSnapshotsTree;
+import org.apache.druid.msq.exec.Controller;
+import org.apache.druid.msq.exec.ControllerClient;
+import org.apache.druid.msq.indexing.MSQTaskList;
+import org.apache.druid.msq.indexing.error.MSQErrorReport;
+import org.apache.druid.msq.kernel.StageId;
+import org.apache.druid.msq.statistics.PartialKeyStatisticsInformation;
+import org.apache.druid.server.security.AuthorizerMapper;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.util.List;
+import java.util.Map;
+
+public class ControllerResource
+{
+ private final Controller controller;
+ private final ResourcePermissionMapper permissionMapper;
+ private final AuthorizerMapper authorizerMapper;
+
+ public ControllerResource(
+ final Controller controller,
+ final ResourcePermissionMapper permissionMapper,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.controller = controller;
+ this.permissionMapper = permissionMapper;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ /**
+ * Used by subtasks to post {@link PartialKeyStatisticsInformation} for shuffling stages.
+ *
+ * See {@link ControllerClient#postPartialKeyStatistics(StageId, int, PartialKeyStatisticsInformation)}
+ * for the client-side code that calls this API.
+ */
+ @POST
+ @Path("/partialKeyStatisticsInformation/{queryId}/{stageNumber}/{workerNumber}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response httpPostPartialKeyStatistics(
+ final Object partialKeyStatisticsObject,
+ @PathParam("queryId") final String queryId,
Review Comment:
## Useless parameter
The parameter 'queryId' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/7152)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/report/MSQResultsReport.java:
##########
@@ -132,9 +72,9 @@
}
@JsonProperty("results")
- public Yielder<Object[]> getResultYielder()
+ public List<Object[]> getResults()
Review Comment:
## Exposing internal representation
getResults exposes the internal representation stored in field results. The value may be modified [after this call to getResults](1).
[Show more details](https://github.com/apache/druid/security/code-scanning/7154)
##########
extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/kernel/controller/MockQueryDefinitionBuilder.java:
##########
@@ -113,27 +126,60 @@
);
definedStages.add(stageNumber);
- ShuffleSpec shuffleSpec;
+ ShuffleSpec shuffleSpec = null;
- if (shuffling) {
- shuffleSpec = new GlobalSortMaxCountShuffleSpec(
- new ClusterBy(
- ImmutableList.of(
- new KeyColumn(SHUFFLE_KEY_COLUMN, KeyOrder.ASCENDING)
+ if (shuffleKind != null) {
+ switch (shuffleKind) {
+ case GLOBAL_SORT:
+ shuffleSpec = new GlobalSortMaxCountShuffleSpec(
+ new ClusterBy(
+ ImmutableList.of(
+ new KeyColumn(SHUFFLE_KEY_COLUMN, KeyOrder.ASCENDING)
+ ),
+ 0
),
- 0
- ),
- MAX_NUM_PARTITIONS,
- false
- );
- } else {
- shuffleSpec = null;
+ MAX_NUM_PARTITIONS,
+ false
+ );
+ break;
+
+ case HASH_LOCAL_SORT:
+ case HASH:
+ shuffleSpec = new HashShuffleSpec(
+ new ClusterBy(
+ ImmutableList.of(
+ new KeyColumn(
+ SHUFFLE_KEY_COLUMN,
+ shuffleKind == ShuffleKind.HASH ? KeyOrder.NONE : KeyOrder.ASCENDING
+ )
+ ),
+ 0
+ ),
+ MAX_NUM_PARTITIONS
+ );
+ break;
+
+ case MIX:
+ shuffleSpec = MixShuffleSpec.instance();
+ break;
+ }
+
+ if (shuffleKind != shuffleSpec.kind()) {
Review Comment:
## Dereferenced variable may be null
Variable [shuffleSpec](1) may be null at this access because of [this](2) assignment.
[Show more details](https://github.com/apache/druid/security/code-scanning/7153)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1573215860
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MemoryIntrospector.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.exec;
+
+import org.apache.druid.msq.kernel.WorkOrder;
+
+/**
+ * Introspector used to generate {@link ControllerMemoryParameters}.
+ */
+public interface MemoryIntrospector
+{
+ /**
+ * Amount of total memory in the entire JVM.
+ */
+ long totalMemoryInJvm();
+
+ /**
+ * Amount of memory usable for the multi-stage query engine in the entire JVM.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long usableMemoryInJvm();
+
+ /**
+ * Amount of total JVM memory required for a particular amount of usable memory to be available.
+ *
+ * This may be an expensive operation. For example, the production implementation {@link MemoryIntrospectorImpl}
+ * estimates size of all lookups as part of computing this value.
+ */
+ long jvmMemoryRequiredForUsableMemory(long usableMemory);
+
+ /**
+ * Maximum number of queries that run simultaneously in this JVM.
+ *
+ * On workers, this is the maximum number of {@link Worker} that run simultaneously. See
+ * {@link WorkerMemoryParameters} for how memory is divided among and within {@link WorkOrder} run by a worker.
Review Comment:
It seems slightly off - a single query can be run simultaneously on multiple tasks. `numQueriesInJvm` should technically be 1, but by the definition in the Javadoc, it might return > 1
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/guice/PeonMemoryManagementModule.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.guice;
+
+import com.google.inject.Binder;
+import com.google.inject.Provides;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.frame.processor.Bouncer;
+import org.apache.druid.guice.LazySingleton;
+import org.apache.druid.guice.annotations.LoadScope;
+import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.msq.exec.MemoryIntrospector;
+import org.apache.druid.msq.exec.MemoryIntrospectorImpl;
+import org.apache.druid.query.lookup.LookupExtractorFactoryContainerProvider;
+import org.apache.druid.utils.JvmUtils;
+
+/**
+ * Provides {@link MemoryIntrospector} for single-task-per-JVM model.
+ *
+ * @see IndexerMemoryManagementModule for multi-task-per-JVM model used on {@link org.apache.druid.cli.CliIndexer}
+ */
+@LoadScope(roles = NodeRole.PEON_JSON_NAME)
+public class PeonMemoryManagementModule implements DruidModule
+{
+ /**
+ * Peons have a single worker per JVM.
+ */
+ private static final int NUM_WORKERS_IN_JVM = 1;
+
+ /**
+ * Peons may have more than one processing thread, but we currently only use one of them.
+ */
+ private static final int NUM_PROCESSING_THREADS = 1;
Review Comment:
I think this hasn't changed from the original MSQ design, but is there any reason why we use a single thread in the Peon? Why are the other threads not utilized?
Also, in the indexer, we use all the threads, but as per my reasoning, we should use a single thread, per process in the indexer (since other processes share that). What's the fallacy in my thinking?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/guice/IndexerMemoryManagementModule.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.guice;
+
+import com.google.inject.Binder;
+import com.google.inject.Provides;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.frame.processor.Bouncer;
+import org.apache.druid.guice.LazySingleton;
+import org.apache.druid.guice.annotations.LoadScope;
+import org.apache.druid.indexing.worker.config.WorkerConfig;
+import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.msq.exec.MemoryIntrospector;
+import org.apache.druid.msq.exec.MemoryIntrospectorImpl;
+import org.apache.druid.query.DruidProcessingConfig;
+import org.apache.druid.query.lookup.LookupExtractorFactoryContainerProvider;
+import org.apache.druid.utils.JvmUtils;
+
+/**
+ * Provides {@link MemoryIntrospector} for multi-task-per-JVM model.
+ *
+ * @see PeonMemoryManagementModule for single-task-per-JVM model used on {@link org.apache.druid.cli.CliPeon}
+ */
+@LoadScope(roles = NodeRole.INDEXER_JSON_NAME)
+public class IndexerMemoryManagementModule implements DruidModule
+{
+ /**
+ * Allocate up to 75% of memory for MSQ-related stuff (if all running tasks are MSQ tasks).
+ */
+ private static final double USABLE_MEMORY_FRACTION = 0.75;
+
+ @Override
+ public void configure(Binder binder)
+ {
+ // Nothing to do.
+ }
+
+ @Provides
+ @LazySingleton
Review Comment:
Since it's a singleton, we should probably rename the method to what resource this bouncer guards. Else there can be confusion as to whether we want to use the global bouncer or create one locally.
Digging into the code, it seems to be guarding the number of the processors that can run concurrently, so something like `makeProcessorBouncer` seems non-ambiguous.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1572844744
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
Review Comment:
nit: ControllerQueryKernelUtils seems appropriate. There are many classes prefixed with Controller.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/StageGroup.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.kernel.StageId;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Group of stages that must be launched as a unit. Within each group, stages communicate with each other using
+ * {@link OutputChannelMode#MEMORY} channels. The final stage in a group writes its own output using
+ * {@link #lastStageOutputChannelMode()}.
+ */
+public class StageGroup
+{
+ private final List<StageId> stageIds;
Review Comment:
It seems that there's an implicit contract that the flow of the data between the stage group is linear (A -> B -> C ...) , and cannot be branched (
A
|
B
| \
C D
| /
E
)
We should probably mention that. Unrelated, but is it subject to change in the future?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/StageGroup.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.kernel.StageId;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Group of stages that must be launched as a unit. Within each group, stages communicate with each other using
+ * {@link OutputChannelMode#MEMORY} channels. The final stage in a group writes its own output using
+ * {@link #lastStageOutputChannelMode()}.
+ */
+public class StageGroup
+{
+ private final List<StageId> stageIds;
+ private final OutputChannelMode groupOutputChannelMode;
+
+ public StageGroup(final List<StageId> stageIds, final OutputChannelMode groupOutputChannelMode)
+ {
+ this.stageIds = stageIds;
+ this.groupOutputChannelMode = groupOutputChannelMode;
+ }
+
+ /**
+ * List of stage IDs in this group.
Review Comment:
As mentioned previously, there's an implicit relation between the stages in the list, and [A, B, C] is not the same as [A, C, B]
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
Review Comment:
nit: Might be cleaner to inverse the condition
```suggestion
if (currentStageId == null) {
continue;
}
```
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
Review Comment:
There's this notion of a group sorting, that is tied to the stage's sorting.
When is a group considered to be sorting, when all, first, or the last of the stages sort? Logically, it should be first, but there's no indication in the code that implies that.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
+ } else {
+ // When the current group doesn't sort, we can't leapfrog.
+ maxStageGroupSizeAllowingForDownstreamConsumer = maxStageGroupSize - 1;
+ }
+
+ currentOutputChannelMode =
+ getOutputChannelMode(
+ queryDef,
+ currentStageId.getStageNumber(),
+ config.getDestination().toSelectDestination(),
+ config.isDurableStorage(),
+ canStream
+
+ // Stages can only stream if they have <= 1 downstream stage (checked by "canStreamOutput") and if
+ // that downstream stage has all of its other inputs available.
+ && (currentOutflow.isEmpty() ||
+ Collections.singleton(currentStageId)
+ .equals(inflow.get(Iterables.getOnlyElement(currentOutflow))))
+
+ // Stages can only stream if that one downstream consumer is able to start concurrently. So, once the
+ // stage group gets too big, we must stop streaming.
+ && (currentOutflow.isEmpty()
+ || currentStageGroup.size() < maxStageGroupSizeAllowingForDownstreamConsumer)
+ );
+
+ currentStageGroup.add(currentStageId);
+
+ if (currentOutflow.size() == 1
+ && currentStageGroup.size() < maxStageGroupSize
+ && currentOutputChannelMode == OutputChannelMode.MEMORY
+
+ // Sorting causes a pipeline break: a sorting stage must read all its input before writing any output.
+ // Continue the stage group only if this stage does not sort its output.
+ && !queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ currentStageId = Iterables.getOnlyElement(currentOutflow);
+ } else {
+ currentStageId = null;
+ }
+ }
+
+ stageGroups.add(new StageGroup(currentStageGroup, currentOutputChannelMode));
+
+ for (final StageId stageId : currentStageGroup) {
+ stagesRun.add(stageId);
+ removeStageFlow(stageId, inflow, outflow);
+ }
+ }
+ }
+
+ return stageGroups;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that flow *into* that stage. Uses TreeMaps and TreeSets so we have a
+ * consistent order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageInflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>()).add(inputStageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that depend on that stage. Uses TreeMaps and TreeSets so we have a consistent
+ * order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(inputStageId, ignored -> new TreeSet<>()).add(stageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Whether output of a stage can possibly use {@link OutputChannelMode#MEMORY}. Returning true does not guarantee
+ * that the stage *will* use {@link OutputChannelMode#MEMORY}. Additional requirements are checked in
+ * {@link #computeStageGroups(QueryDefinition, ControllerQueryKernelConfig)}.
+ */
+ public static boolean canStreamOutput(
+ final QueryDefinition queryDefinition,
+ final int stageNumber,
+ final ControllerQueryKernelConfig config,
+ final Map<StageId, Set<StageId>> outflowMap
+ )
+ {
+ if (config.isFaultTolerant()) {
+ // Cannot stream if fault tolerance is enabled: durable storage is required.
+ return false;
+ }
+
+ if (!config.isPipeline() || config.getMaxConcurrentStages() < 2) {
+ // Cannot stream if pipelining (& running multiple stages at once) is not enabled.
+ return false;
+ }
+
+ final StageId stageId = queryDefinition.getStageDefinition(stageNumber).getId();
+ final Set<StageId> outflowStageIds = outflowMap.get(stageId);
+
+ if (outflowStageIds.isEmpty()) {
+ return true;
+ } else if (outflowStageIds.size() == 1) {
+ final StageDefinition outflowStageDef =
+ queryDefinition.getStageDefinition(Iterables.getOnlyElement(outflowStageIds));
+
+ // Two things happening here:
+ // 1) Stages cannot both stream and broadcast their output. This is because when streaming, we can only
+ // support a single reader.
+ // 2) Stages can only receive a single streamed input. This isn't strictly necessary, but it simplifies the
+ // logic around concurrently launching stages.
+ return stageId.equals(getOnlyNonBroadcastInputAsStageId(outflowStageDef));
+ } else {
+ return false;
+ }
+ }
+
+ /**
+ * Return an {@link OutputChannelMode} to use for a given stage, based on query and context.
+ */
+ public static OutputChannelMode getOutputChannelMode(
+ final QueryDefinition queryDef,
+ final int stageNumber,
+ @Nullable final MSQSelectDestination selectDestination,
+ final boolean durableStorage,
+ final boolean canStream
+ )
+ {
+ final boolean isFinalStage = queryDef.getFinalStageDefinition().getStageNumber() == stageNumber;
+
+ if (isFinalStage && selectDestination == MSQSelectDestination.DURABLESTORAGE) {
+ return OutputChannelMode.DURABLE_STORAGE_QUERY_RESULTS;
+ } else if (canStream) {
+ return OutputChannelMode.MEMORY;
+ } else if (durableStorage) {
+ return OutputChannelMode.DURABLE_STORAGE_INTERMEDIATE;
+ } else {
+ return OutputChannelMode.LOCAL_STORAGE;
+ }
+ }
+
+ /**
+ * If a stage has a single non-broadcast input stage, returns that input stage. Otherwise, returns null.
+ * This is a helper used by {@link #canStreamOutput}.
+ */
+ @Nullable
+ public static StageId getOnlyNonBroadcastInputAsStageId(final StageDefinition downstreamStageDef)
+ {
+ final List<InputSpec> inputSpecs = downstreamStageDef.getInputSpecs();
+ final IntSet broadcastInputNumbers = downstreamStageDef.getBroadcastInputNumbers();
+
+ if (inputSpecs.size() - broadcastInputNumbers.size() != 1) {
+ return null;
+ }
+
+ for (int i = 0; i < inputSpecs.size(); i++) {
+ if (!broadcastInputNumbers.contains(i)) {
+ final IntSet stageNumbers = InputSpecs.getStageNumbers(Collections.singletonList(inputSpecs.get(i)));
+ if (stageNumbers.size() == 1) {
+ return new StageId(downstreamStageDef.getId().getQueryId(), stageNumbers.iterator().nextInt());
+ }
+ }
+ }
+
+ return null;
+ }
+
+ private static void removeStageFlow(
Review Comment:
nit: Javadoc about preconditions and postconditions of the method, and what modifications it makes to the input maps.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
Review Comment:
What's the significance of inflow being empty?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
Review Comment:
There's an implicit assumption that the inflow's keySet is sorted, which is true since it is actually a TreeMap, however isn't reflected here. Perhaps we should tighten the contract that the inflow map is a TreeMap, instead of a Map (which can be a HashMap as well, which wouldn't work well with this logic)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
Review Comment:
Doesn't this defy the logic of stage groups - each group runs individually, and all the stages within a group run simultanously. If the prior group runs concurrently with this group, shouldn't they be coalesced into a single group?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
Review Comment:
Can this ever be 0, and break the logic. Partially related to the question posed above.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
+ } else {
+ // When the current group doesn't sort, we can't leapfrog.
+ maxStageGroupSizeAllowingForDownstreamConsumer = maxStageGroupSize - 1;
+ }
+
+ currentOutputChannelMode =
+ getOutputChannelMode(
+ queryDef,
+ currentStageId.getStageNumber(),
+ config.getDestination().toSelectDestination(),
+ config.isDurableStorage(),
+ canStream
+
+ // Stages can only stream if they have <= 1 downstream stage (checked by "canStreamOutput") and if
+ // that downstream stage has all of its other inputs available.
+ && (currentOutflow.isEmpty() ||
+ Collections.singleton(currentStageId)
+ .equals(inflow.get(Iterables.getOnlyElement(currentOutflow))))
+
+ // Stages can only stream if that one downstream consumer is able to start concurrently. So, once the
+ // stage group gets too big, we must stop streaming.
+ && (currentOutflow.isEmpty()
+ || currentStageGroup.size() < maxStageGroupSizeAllowingForDownstreamConsumer)
+ );
+
+ currentStageGroup.add(currentStageId);
+
+ if (currentOutflow.size() == 1
+ && currentStageGroup.size() < maxStageGroupSize
+ && currentOutputChannelMode == OutputChannelMode.MEMORY
+
+ // Sorting causes a pipeline break: a sorting stage must read all its input before writing any output.
+ // Continue the stage group only if this stage does not sort its output.
+ && !queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ currentStageId = Iterables.getOnlyElement(currentOutflow);
+ } else {
+ currentStageId = null;
+ }
+ }
+
+ stageGroups.add(new StageGroup(currentStageGroup, currentOutputChannelMode));
+
+ for (final StageId stageId : currentStageGroup) {
+ stagesRun.add(stageId);
+ removeStageFlow(stageId, inflow, outflow);
+ }
+ }
+ }
+
+ return stageGroups;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that flow *into* that stage. Uses TreeMaps and TreeSets so we have a
+ * consistent order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageInflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>()).add(inputStageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Returns a mapping of stage -> stages that depend on that stage. Uses TreeMaps and TreeSets so we have a consistent
+ * order for analyzing and running stages.
+ */
+ public static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefinition queryDefinition)
+ {
+ final Map<StageId, Set<StageId>> retVal = new TreeMap<>();
+
+ for (final StageDefinition stageDef : queryDefinition.getStageDefinitions()) {
+ final StageId stageId = stageDef.getId();
+ retVal.computeIfAbsent(stageId, ignored -> new TreeSet<>());
+
+ for (final int inputStageNumber : queryDefinition.getStageDefinition(stageId).getInputStageNumbers()) {
+ final StageId inputStageId = new StageId(queryDefinition.getQueryId(), inputStageNumber);
+ retVal.computeIfAbsent(inputStageId, ignored -> new TreeSet<>()).add(stageId);
+ }
+ }
+
+ return retVal;
+ }
+
+ /**
+ * Whether output of a stage can possibly use {@link OutputChannelMode#MEMORY}. Returning true does not guarantee
+ * that the stage *will* use {@link OutputChannelMode#MEMORY}. Additional requirements are checked in
+ * {@link #computeStageGroups(QueryDefinition, ControllerQueryKernelConfig)}.
+ */
+ public static boolean canStreamOutput(
+ final QueryDefinition queryDefinition,
+ final int stageNumber,
+ final ControllerQueryKernelConfig config,
+ final Map<StageId, Set<StageId>> outflowMap
+ )
+ {
+ if (config.isFaultTolerant()) {
+ // Cannot stream if fault tolerance is enabled: durable storage is required.
+ return false;
+ }
+
+ if (!config.isPipeline() || config.getMaxConcurrentStages() < 2) {
+ // Cannot stream if pipelining (& running multiple stages at once) is not enabled.
+ return false;
+ }
+
+ final StageId stageId = queryDefinition.getStageDefinition(stageNumber).getId();
+ final Set<StageId> outflowStageIds = outflowMap.get(stageId);
+
+ if (outflowStageIds.isEmpty()) {
+ return true;
+ } else if (outflowStageIds.size() == 1) {
+ final StageDefinition outflowStageDef =
+ queryDefinition.getStageDefinition(Iterables.getOnlyElement(outflowStageIds));
+
+ // Two things happening here:
+ // 1) Stages cannot both stream and broadcast their output. This is because when streaming, we can only
+ // support a single reader.
+ // 2) Stages can only receive a single streamed input. This isn't strictly necessary, but it simplifies the
Review Comment:
What is the signifcance of streamed input? There is probably some gap in my understanding of streamed, but would we call the data lying in durable storage ready to be read as streamed? Logically the stage can stream the output in that case, and per my understanding, this code would allow that to happen, but I wouldn't call it "streamed" input.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.druid.msq.kernel.controller;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.msq.exec.OutputChannelMode;
+import org.apache.druid.msq.indexing.destination.MSQDestination;
+import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
+import org.apache.druid.msq.input.InputSpec;
+import org.apache.druid.msq.input.InputSpecs;
+import org.apache.druid.msq.kernel.QueryDefinition;
+import org.apache.druid.msq.kernel.StageDefinition;
+import org.apache.druid.msq.kernel.StageId;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+/**
+ * Utilties for {@link ControllerQueryKernel}.
+ */
+public class ControllerUtils
+{
+ /**
+ * Put stages from {@link QueryDefinition} into groups that must each be launched simultaneously.
+ *
+ * This method's goal is to maximize the usage of {@link OutputChannelMode#MEMORY} channels, subject to constraints
+ * provided by {@link ControllerQueryKernelConfig#isPipeline()},
+ * {@link ControllerQueryKernelConfig#getMaxConcurrentStages()}, and
+ * {@link ControllerQueryKernelConfig#isFaultTolerant()}.
+ */
+ public static List<StageGroup> computeStageGroups(
+ final QueryDefinition queryDef,
+ final ControllerQueryKernelConfig config
+ )
+ {
+ final MSQDestination destination = config.getDestination();
+ final List<StageGroup> stageGroups = new ArrayList<>();
+ final boolean useDurableStorage = config.isDurableStorage();
+ final Map<StageId, Set<StageId>> inflow = computeStageInflowMap(queryDef);
+ final Map<StageId, Set<StageId>> outflow = computeStageOutflowMap(queryDef);
+ final Set<StageId> stagesRun = new HashSet<>();
+
+ while (stagesRun.size() < queryDef.getStageDefinitions().size()) {
+ // 1) Run all stages that cannot stream their output, as solo groups.
+ boolean didRun;
+ do {
+ didRun = false;
+
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && !canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ stagesRun.add(stageId);
+ stageGroups.add(
+ new StageGroup(
+ Collections.singletonList(stageId),
+ getOutputChannelMode(
+ queryDef,
+ stageId.getStageNumber(),
+ destination.toSelectDestination(),
+ useDurableStorage,
+ false
+ )
+ )
+ );
+
+ removeStageFlow(stageId, inflow, outflow);
+ didRun = true;
+ }
+ }
+ } while (didRun);
+
+ // 2) Pick some stage that can stream its output, and run that as well as all ready-to-run dependents.
+ StageId currentStageId = null;
+ for (final StageId stageId : ImmutableList.copyOf(inflow.keySet())) {
+ if (!stagesRun.contains(stageId)
+ && inflow.get(stageId).isEmpty()
+ && canStreamOutput(queryDef, stageId.getStageNumber(), config, outflow)) {
+ currentStageId = stageId;
+ break;
+ }
+ }
+
+ if (currentStageId != null) {
+ final List<StageId> currentStageGroup = new ArrayList<>();
+ final int maxStageGroupSize;
+ if (stageGroups.isEmpty()) {
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ } else {
+ final StageGroup priorGroup = stageGroups.get(stageGroups.size() - 1);
+ if (priorGroup.lastStageOutputChannelMode() == OutputChannelMode.MEMORY) {
+ // Prior group must run concurrently with this group.
+ maxStageGroupSize = config.getMaxConcurrentStages() - priorGroup.size();
+ } else {
+ // Prior group can exit before this group starts.
+ maxStageGroupSize = config.getMaxConcurrentStages();
+ }
+ }
+
+ OutputChannelMode currentOutputChannelMode = null;
+ while (currentStageId != null) {
+ final boolean canStream = canStreamOutput(queryDef, currentStageId.getStageNumber(), config, outflow);
+ final Set<StageId> currentOutflow = outflow.get(currentStageId);
+
+ final int maxStageGroupSizeAllowingForDownstreamConsumer;
+ if (queryDef.getStageDefinition(currentStageId).doesSortDuringShuffle()) {
+ // When the current group sorts, there's a pipeline break, so we can "leapfrog": close the prior group
+ // before starting the downstream group.
+ maxStageGroupSizeAllowingForDownstreamConsumer = config.getMaxConcurrentStages() - 1;
Review Comment:
Can something like following happen:
a) Prior's group ouptut type is OutputChannelMode.MEMORY (and prior + this group must run together)
b) The group sorts, therefore 'maxStageGroupSizeAllowingForDownstreamConsumer' = maxConcurrentStages - 1
c) The actual number of stages running simultaneously will be (priorGroup.size() + maxConcurrentStages - 1), which can be greater than maxConcurrentStages.
If not, what's there to prevent it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ controller: Support in-memory shuffles; towards JVM reuse. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16168:
URL: https://github.com/apache/druid/pull/16168#discussion_r1558117791
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -2995,15 +2873,14 @@
*/
private interface TaskContactFn
{
- ListenableFuture<Void> contactTask(WorkerClient client, String taskId, int workerNumber);
+ ListenableFuture<Void> contactTask(WorkerClient client, String workerId, int workerNumber);
}
/**
* Interface used when {@link TaskContactFn#contactTask(WorkerClient, String, int)} returns a successful future.
*/
private interface TaskContactSuccess
{
- void onSuccess(String taskId);
-
+ void onSuccess(String workerId, int workerNumber);
Review Comment:
I meant to use this in a log message; added the log message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org