Skip to content

Commit 8d959a0

Browse files
HanumathRaoHanumathRao
committed
Integration changes with new DistributedRM queue configuration.
a) Remove the redundant NodeResource and merge the additional member functions with the NodeResources class. b) Added new UUID logic and selection of a queue based on the memory requirement during parallelization phase. c) Changed proto definitions to set the UUID of a drillbit. d) Implementation of new DrillNode Wrapper over DrillbitEndpoint to fix the equality comparisions between DrillbitEndpoints.
1 parent 21098e3 commit 8d959a0

24 files changed

+494
-199
lines changed

‎exec/java-exec/src/main/java/org/apache/drill/common/DrillNode.java

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.drill.common;
19+
20+
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
21+
22+
/**
23+
* DrillNode encapsulates a drillendpoint. DrillbitEndpoint is a protobuf generated class which requires
24+
* all the member variables to be equal for DrillbitEndpoints to be equal. DrillNode relaxes this requirement
25+
* by only comparing required variables.
26+
*/
27+
public class DrillNode {
28+
private final DrillbitEndpoint endpoint;
29+
30+
public DrillNode(DrillbitEndpoint endpoint) {
31+
this.endpoint = endpoint;
32+
}
33+
34+
public static DrillNode create(DrillbitEndpoint endpoint) {
35+
return new DrillNode(endpoint);
36+
}
37+
38+
public boolean equals(Object other) {
39+
if (!(other instanceof DrillNode)) {
40+
return false;
41+
}
42+
43+
DrillbitEndpoint otherEndpoint = ((DrillNode) other).endpoint;
44+
return endpoint.getAddress().equals(otherEndpoint.getAddress()) &&
45+
endpoint.getUserPort() == otherEndpoint.getUserPort() &&
46+
endpoint.getControlPort() == otherEndpoint.getControlPort() &&
47+
endpoint.getDataPort() == otherEndpoint.getDataPort() &&
48+
endpoint.getVersion().equals(otherEndpoint.getVersion());
49+
}
50+
51+
@Override
52+
public int hashCode() {
53+
int hash = 41;
54+
hash = (19 * hash) + endpoint.getDescriptor().hashCode();
55+
if (endpoint.hasAddress()) {
56+
hash = (37 * hash) + endpoint.ADDRESS_FIELD_NUMBER;
57+
hash = (53 * hash) + endpoint.getAddress().hashCode();
58+
}
59+
if (endpoint.hasUserPort()) {
60+
hash = (37 * hash) + endpoint.USER_PORT_FIELD_NUMBER;
61+
hash = (53 * hash) + endpoint.getUserPort();
62+
}
63+
if (endpoint.hasControlPort()) {
64+
hash = (37 * hash) + endpoint.CONTROL_PORT_FIELD_NUMBER;
65+
hash = (53 * hash) + endpoint.getControlPort();
66+
}
67+
if (endpoint.hasDataPort()) {
68+
hash = (37 * hash) + endpoint.DATA_PORT_FIELD_NUMBER;
69+
hash = (53 * hash) + endpoint.getDataPort();
70+
}
71+
if (endpoint.hasVersion()) {
72+
hash = (37 * hash) + endpoint.VERSION_FIELD_NUMBER;
73+
hash = (53 * hash) + endpoint.getVersion().hashCode();
74+
}
75+
return hash;
76+
}
77+
78+
79+
@Override
80+
public String toString() {
81+
StringBuilder sb = new StringBuilder();
82+
83+
return sb.append("endpoint address :")
84+
.append(endpoint.getAddress())
85+
.append("endpoint user port: ")
86+
.append(endpoint.getUserPort()).toString();
87+
}
88+
}

‎exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,10 @@
1717
*/
1818
package org.apache.drill.exec.ops;
1919

20-
import java.util.Collection;
21-
import java.util.List;
22-
import java.util.Map;
23-
20+
import io.netty.buffer.DrillBuf;
2421
import org.apache.calcite.schema.SchemaPlus;
2522
import org.apache.drill.common.AutoCloseables;
23+
import org.apache.drill.common.DrillNode;
2624
import org.apache.drill.common.config.DrillConfig;
2725
import org.apache.drill.common.config.LogicalPlanPersistence;
2826
import org.apache.drill.common.types.TypeProtos.MinorType;
@@ -51,12 +49,13 @@
5149
import org.apache.drill.exec.store.StoragePluginRegistry;
5250
import org.apache.drill.exec.testing.ExecutionControls;
5351
import org.apache.drill.exec.util.Utilities;
54-
5552
import org.apache.drill.shaded.guava.com.google.common.base.Function;
5653
import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
5754
import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
5855

59-
import io.netty.buffer.DrillBuf;
56+
import java.util.Collection;
57+
import java.util.List;
58+
import java.util.Map;
6059

6160
// TODO - consider re-name to PlanningContext, as the query execution context actually appears
6261
// in fragment contexts
@@ -242,6 +241,14 @@ public Collection<DrillbitEndpoint> getOnlineEndpoints() {
242241
return drillbitContext.getBits();
243242
}
244243

244+
/**
245+
* TODO: Change it to use {@link DrillNode} instead of DrillbitEndpoint
246+
* @return map of endpoint to UUIDs
247+
*/
248+
public Map<DrillbitEndpoint, String> getOnlineEndpointUUIDs() {
249+
return drillbitContext.getOnlineEndpointUUIDs();
250+
}
251+
245252
public DrillConfig getConfig() {
246253
return drillbitContext.getConfig();
247254
}

‎exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/NodeResource.java

Lines changed: 0 additions & 73 deletions
This file was deleted.

‎exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/DefaultParallelizer.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@
2222
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
2323
import org.apache.drill.exec.util.memory.DefaultMemoryAllocationUtilities;
2424

25-
import java.util.Collection;
2625
import java.util.List;
26+
import java.util.Map;
2727
import java.util.Set;
2828
import java.util.function.BiFunction;
2929

@@ -51,7 +51,7 @@ public DefaultParallelizer(boolean memoryPlanning, long parallelizationThreshold
5151

5252
@Override
5353
public void adjustMemory(PlanningSet planningSet, Set<Wrapper> roots,
54-
Collection<DrillbitEndpoint> activeEndpoints) {
54+
Map<DrillbitEndpoint, String> onlineEndpointUUIDs) {
5555
if (planHasMemory) {
5656
return;
5757
}

‎exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/DistributedQueueParallelizer.java

Lines changed: 51 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,16 @@
1818
package org.apache.drill.exec.planner.fragment;
1919

2020
import org.apache.commons.lang3.tuple.Pair;
21+
import org.apache.drill.common.exceptions.ExecutionSetupException;
2122
import org.apache.drill.common.util.function.CheckedConsumer;
2223
import org.apache.drill.exec.ops.QueryContext;
2324
import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
2425
import org.apache.drill.exec.physical.base.PhysicalOperator;
25-
import org.apache.drill.exec.planner.cost.NodeResource;
2626
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
27+
import org.apache.drill.exec.resourcemgr.NodeResources;
28+
import org.apache.drill.exec.resourcemgr.config.QueryQueueConfig;
29+
import org.apache.drill.exec.resourcemgr.config.exception.QueueSelectionException;
30+
import org.apache.drill.exec.work.foreman.rm.QueryResourceManager;
2731

2832
import java.util.Map;
2933
import java.util.HashMap;
@@ -43,12 +47,14 @@
4347
public class DistributedQueueParallelizer extends SimpleParallelizer {
4448
private final boolean planHasMemory;
4549
private final QueryContext queryContext;
50+
private final QueryResourceManager rm;
4651
private final Map<DrillbitEndpoint, Map<PhysicalOperator, Long>> operators;
4752

48-
public DistributedQueueParallelizer(boolean memoryPlanning, QueryContext queryContext) {
53+
public DistributedQueueParallelizer(boolean memoryPlanning, QueryContext queryContext, QueryResourceManager queryRM) {
4954
super(queryContext);
5055
this.planHasMemory = memoryPlanning;
5156
this.queryContext = queryContext;
57+
this.rm = queryRM;
5258
this.operators = new HashMap<>();
5359
}
5460

@@ -75,45 +81,75 @@ public BiFunction<DrillbitEndpoint, PhysicalOperator, Long> getMemory() {
7581
*
7682
* @param planningSet context of the fragments.
7783
* @param roots root fragments.
78-
* @param activeEndpoints currently active endpoints.
84+
* @param onlineEndpointUUIDs currently active endpoints.
7985
* @throws PhysicalOperatorSetupException
8086
*/
8187
public void adjustMemory(PlanningSet planningSet, Set<Wrapper> roots,
82-
Collection<DrillbitEndpoint> activeEndpoints) throws PhysicalOperatorSetupException {
88+
Map<DrillbitEndpoint, String> onlineEndpointUUIDs) throws ExecutionSetupException {
8389

8490
if (planHasMemory) {
8591
return;
8692
}
8793
// total node resources for the query plan maintained per drillbit.
88-
final Map<DrillbitEndpoint, NodeResource> totalNodeResources =
89-
activeEndpoints.stream().collect(Collectors.toMap(x ->x,
90-
x -> NodeResource.create()));
94+
final Map<DrillbitEndpoint, NodeResources> totalNodeResources =
95+
onlineEndpointUUIDs.keySet().stream().collect(Collectors.toMap(x ->x,
96+
x -> NodeResources.create()));
9197

9298
// list of the physical operators and their memory requirements per drillbit.
9399
final Map<DrillbitEndpoint, List<Pair<PhysicalOperator, Long>>> operators =
94-
activeEndpoints.stream().collect(Collectors.toMap(x -> x,
100+
onlineEndpointUUIDs.keySet().stream().collect(Collectors.toMap(x -> x,
95101
x -> new ArrayList<>()));
96102

97103
for (Wrapper wrapper : roots) {
98104
traverse(wrapper, CheckedConsumer.throwingConsumerWrapper((Wrapper fragment) -> {
99105
MemoryCalculator calculator = new MemoryCalculator(planningSet, queryContext);
100106
fragment.getNode().getRoot().accept(calculator, fragment);
101-
NodeResource.merge(totalNodeResources, fragment.getResourceMap());
107+
NodeResources.merge(totalNodeResources, fragment.getResourceMap());
102108
operators.entrySet()
103109
.stream()
104110
.forEach((entry) -> entry.getValue()
105111
.addAll(calculator.getBufferedOperators(entry.getKey())));
106112
}));
107113
}
108-
//queryrm.selectQueue( pass the max node Resource) returns queue configuration.
109-
Map<DrillbitEndpoint, List<Pair<PhysicalOperator, Long>>> memoryAdjustedOperators = ensureOperatorMemoryWithinLimits(operators, totalNodeResources, 10);
114+
115+
QueryQueueConfig queueConfig = null;
116+
try {
117+
queueConfig = this.rm.selectQueue(max(totalNodeResources.values()));
118+
} catch (QueueSelectionException exception) {
119+
throw new ExecutionSetupException(exception.getMessage());
120+
}
121+
122+
Map<DrillbitEndpoint,
123+
List<Pair<PhysicalOperator, Long>>> memoryAdjustedOperators = ensureOperatorMemoryWithinLimits(operators, totalNodeResources,
124+
queueConfig.getMaxQueryMemoryInMBPerNode());
110125
memoryAdjustedOperators.entrySet().stream().forEach((x) -> {
111126
Map<PhysicalOperator, Long> memoryPerOperator = x.getValue().stream()
112127
.collect(Collectors.toMap(operatorLongPair -> operatorLongPair.getLeft(),
113128
operatorLongPair -> operatorLongPair.getRight(),
114129
(mem_1, mem_2) -> (mem_1 + mem_2)));
115130
this.operators.put(x.getKey(), memoryPerOperator);
116131
});
132+
133+
this.rm.setCost(convertToUUID(totalNodeResources, onlineEndpointUUIDs));
134+
}
135+
136+
private Map<String, NodeResources> convertToUUID(Map<DrillbitEndpoint, NodeResources> nodeResourcesMap,
137+
Map<DrillbitEndpoint, String> onlineEndpointUUIDs) {
138+
Map<String, NodeResources> nodeResourcesPerUUID = new HashMap<>();
139+
for (Map.Entry<DrillbitEndpoint, NodeResources> nodeResource : nodeResourcesMap.entrySet()) {
140+
nodeResourcesPerUUID.put(onlineEndpointUUIDs.get(nodeResource.getKey()), nodeResource.getValue());
141+
}
142+
return nodeResourcesPerUUID;
143+
}
144+
145+
private NodeResources max(Collection<NodeResources> resources) {
146+
NodeResources maxResource = null;
147+
for (NodeResources resource : resources) {
148+
if (maxResource == null || maxResource.getMemoryInBytes() < resource.getMemoryInBytes()) {
149+
maxResource = resource;
150+
}
151+
}
152+
return maxResource;
117153
}
118154

119155

@@ -126,12 +162,12 @@ public void adjustMemory(PlanningSet planningSet, Set<Wrapper> roots,
126162
*/
127163
private Map<DrillbitEndpoint, List<Pair<PhysicalOperator, Long>>>
128164
ensureOperatorMemoryWithinLimits(Map<DrillbitEndpoint, List<Pair<PhysicalOperator, Long>>> memoryPerOperator,
129-
Map<DrillbitEndpoint, NodeResource> nodeResourceMap, int nodeLimit) {
165+
Map<DrillbitEndpoint, NodeResources> nodeResourceMap, long nodeLimit) {
130166
// Get the physical operators which are above the node memory limit.
131167
Map<DrillbitEndpoint, List<Pair<PhysicalOperator, Long>>> onlyMemoryAboveLimitOperators = new HashMap<>();
132168
memoryPerOperator.entrySet().stream().forEach((entry) -> {
133169
onlyMemoryAboveLimitOperators.putIfAbsent(entry.getKey(), new ArrayList<>());
134-
if (nodeResourceMap.get(entry.getKey()).getMemory() > nodeLimit) {
170+
if (nodeResourceMap.get(entry.getKey()).getMemoryInBytes() > nodeLimit) {
135171
onlyMemoryAboveLimitOperators.get(entry.getKey()).addAll(entry.getValue());
136172
}
137173
});
@@ -148,6 +184,8 @@ public void adjustMemory(PlanningSet planningSet, Set<Wrapper> roots,
148184
return Pair.of(operatorMemory.getKey(), (long) Math.ceil(operatorMemory.getValue()/totalMemory * nodeLimit));
149185
}).collect(Collectors.toList());
150186
memoryAdjustedDrillbits.put(entry.getKey(), adjustedMemory);
187+
NodeResources nodeResources = nodeResourceMap.get(entry.getKey());
188+
nodeResources.setMemoryInBytes(adjustedMemory.stream().mapToLong(Pair::getValue).sum());
151189
}
152190
);
153191

‎exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/MemoryCalculator.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,8 @@
2323
import org.apache.drill.exec.physical.base.PhysicalOperator;
2424
import org.apache.drill.exec.physical.config.AbstractMuxExchange;
2525
import org.apache.drill.exec.planner.AbstractOpWrapperVisitor;
26-
import org.apache.drill.exec.planner.cost.NodeResource;
2726
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
27+
import org.apache.drill.exec.resourcemgr.NodeResources;
2828
import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
2929

3030
import java.util.ArrayList;
@@ -70,11 +70,11 @@ private void merge(Wrapper currFrag,
7070
Map<DrillbitEndpoint, Integer> minorFragsPerDrillBit,
7171
Function<Entry<DrillbitEndpoint, Integer>, Long> getMemory) {
7272

73-
NodeResource.merge(currFrag.getResourceMap(),
73+
NodeResources.merge(currFrag.getResourceMap(),
7474
minorFragsPerDrillBit.entrySet()
7575
.stream()
7676
.collect(Collectors.toMap((x) -> x.getKey(),
77-
(x) -> NodeResource.create(0,
77+
(x) -> NodeResources.create(0,
7878
getMemory.apply(x)))));
7979
}
8080

0 commit comments

Comments
 (0)