[ https://issues.apache.org/jira/browse/DRILL-4706?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15638057#comment-15638057 ]
ASF GitHub Bot commented on DRILL-4706: --------------------------------------- Github user sohami commented on a diff in the pull request: https://github.com/apache/drill/pull/639#discussion_r86605131 --- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/LocalAffinityFragmentParallelizer.java --- @@ -0,0 +1,165 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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.drill.exec.planner.fragment; + +import com.google.common.collect.Lists; +import com.google.common.collect.Ordering; +import org.apache.drill.exec.physical.EndpointAffinity; +import org.apache.drill.exec.physical.PhysicalOperatorSetupException; +import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint; + +import java.util.Map; +import java.util.List; +import java.util.Collection; +import java.util.HashMap; +import java.util.Collections; + +/** + * Implementation of {@link FragmentParallelizer} where fragment has zero or more endpoints. + * This is for Parquet Scan Fragments only. Fragment placement is done preferring + * data locality. + */ +public class LocalAffinityFragmentParallelizer implements FragmentParallelizer { + public static final LocalAffinityFragmentParallelizer INSTANCE = new LocalAffinityFragmentParallelizer(); + + // Sort a list of map entries by values. + Ordering<Map.Entry<DrillbitEndpoint, Integer>> sortByValues = new Ordering<Map.Entry<DrillbitEndpoint, Integer>>() { + @Override + public int compare(Map.Entry<DrillbitEndpoint, Integer> left, Map.Entry<DrillbitEndpoint, Integer> right) { + return right.getValue().compareTo(left.getValue()); + } + }; + + @Override + public void parallelizeFragment(final Wrapper fragmentWrapper, final ParallelizationParameters parameters, + final Collection<DrillbitEndpoint> activeEndpoints) throws PhysicalOperatorSetupException { + final Stats stats = fragmentWrapper.getStats(); + final ParallelizationInfo parallelizationInfo = stats.getParallelizationInfo(); + final Map<DrillbitEndpoint, EndpointAffinity> endpointAffinityMap = + fragmentWrapper.getStats().getParallelizationInfo().getEndpointAffinityMap(); + int totalWorkUnits = 0; + Map<DrillbitEndpoint, Integer> endpointPool = new HashMap<>(); + + // Get the total number of work units and list of endPoints to schedule fragments on + for (Map.Entry<DrillbitEndpoint, EndpointAffinity> epAff : endpointAffinityMap.entrySet()) { + if (epAff.getValue().getNumLocalWorkUnits() > 0) { + totalWorkUnits += epAff.getValue().getNumLocalWorkUnits(); + endpointPool.put(epAff.getKey(), epAff.getValue().getNumLocalWorkUnits()); + } + } + + // Find the parallelization width of fragment + // 1. Find the parallelization based on cost. Use max cost of all operators in this fragment; this is consistent + // with the calculation that ExcessiveExchangeRemover uses. + int width = (int) Math.ceil(stats.getMaxCost() / parameters.getSliceTarget()); + + // 2. Cap the parallelization width by fragment level width limit and system level per query width limit + width = Math.min(width, Math.min(parallelizationInfo.getMaxWidth(), parameters.getMaxGlobalWidth())); + + // 3. Cap the parallelization width by system level per node width limit + width = Math.min(width, parameters.getMaxWidthPerNode() * endpointPool.size()); + + // 4. Make sure width is at least the min width enforced by operators + width = Math.max(parallelizationInfo.getMinWidth(), width); + + // 5. Make sure width is at most the max width enforced by operators + width = Math.min(parallelizationInfo.getMaxWidth(), width); + + // 6: Finally make sure the width is at least one + width = Math.max(1, width); + + List<DrillbitEndpoint> assignedEndPoints = Lists.newArrayList(); + int totalAssigned = 0; + + // Sort the endpointPool based on numLocalWorkUnits. This sorting is done because we are doing + // round robin allocation and we stop when we reach the width. We want to allocate + // on endpoints which have higher numLocalWorkUnits first. + List<Map.Entry<DrillbitEndpoint, Integer>> sortedEndpointPool = Lists.newArrayList(endpointPool.entrySet()); + Collections.sort(sortedEndpointPool, sortByValues); + + // Keep track of number of fragments allocated to each endpoint. + Map<DrillbitEndpoint, Integer> endpointAssignments = new HashMap<>(); + + // Keep track of how many more to assign to each endpoint. + Map<DrillbitEndpoint, Integer> remainingEndpointAssignments = new HashMap<>(); + + // Calculate the target allocation for each endPoint based on work it has to do + // Assign one fragment (minimum) to all the endPoints in the pool. + for (DrillbitEndpoint ep : endpointPool.keySet()) { + final int targetAllocation = (int) Math.ceil(endpointAffinityMap.get(ep).getNumLocalWorkUnits() * ((double)width / parallelizationInfo.getMaxWidth())); + assignedEndPoints.add(ep); + totalAssigned++; + remainingEndpointAssignments.put(ep, targetAllocation-1); + endpointAssignments.put(ep, 1); + } + + // Keep allocating from endpoints in a round robin fashion upto + // max(targetAllocation, maxwidthPerNode) for each endpoint and --- End diff -- Wrong comment.. We assign until we reach limit of maxWidthPerNode > Fragment planning causes Drillbits to read remote chunks when local copies > are available > ---------------------------------------------------------------------------------------- > > Key: DRILL-4706 > URL: https://issues.apache.org/jira/browse/DRILL-4706 > Project: Apache Drill > Issue Type: Bug > Components: Query Planning & Optimization > Affects Versions: 1.6.0 > Environment: CentOS, RHEL > Reporter: Kunal Khatua > Assignee: Sorabh Hamirwasia > Labels: performance, planning > > When a table (datasize=70GB) of 160 parquet files (each having a single > rowgroup and fitting within one chunk) is available on a 10-node setup with > replication=3 ; a pure data scan query causes about 2% of the data to be read > remotely. > Even with the creation of metadata cache, the planner is selecting a > sub-optimal plan of executing the SCAN fragments such that some of the data > is served from a remote server. -- This message was sent by Atlassian JIRA (v6.3.4#6332)