Package org.apache.drill.exec.planner.fragment

Class SimpleParallelizer

java.lang.Object

org.apache.drill.exec.planner.fragment.SimpleParallelizer

All Implemented Interfaces:

ParallelizationParameters, QueryParallelizer

Direct Known Subclasses:

DefaultQueryParallelizer, QueueQueryParallelizer

public abstract class SimpleParallelizer
extends Object
implements QueryParallelizer

The simple parallelizer determines the level of parallelization of a plan based on the cost of the underlying operations. It doesn't take into account system load or other factors. Based on the cost of the query, the parallelization for each major fragment will be determined. Once the amount of parallelization is done, assignment is done based on round robin assignment ordered by operator affinity (locality) to available execution Drillbits.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected static class	SimpleParallelizer.CountRequiredFragments	Designed to setup initial values for arriving fragment accounting.

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	SimpleParallelizer(long parallelizationThreshold, int maxWidthPerNode, int maxGlobalWidth, double affinityFactor)
protected	SimpleParallelizer(QueryContext context)

Method Summary

Modifier and Type	Method	Description
abstract void	adjustMemory(PlanningSet planningSet, Set<Wrapper> roots, Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints)
void	collectStatsAndParallelizeFragments(PlanningSet planningSet, Set<Wrapper> roots, Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints)	Traverse all the major fragments and parallelize each major fragment based on collected stats.
final QueryWorkUnit	generateWorkUnit(OptionList options, CoordinationProtos.DrillbitEndpoint foremanNode, UserBitShared.QueryId queryId, Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints, Fragment rootFragment, UserSession session, BitControl.QueryContextInformation queryContextInfo)	The starting function for the whole parallelization and memory computation logic.
protected QueryWorkUnit	generateWorkUnit(OptionList options, CoordinationProtos.DrillbitEndpoint foremanNode, UserBitShared.QueryId queryId, Fragment rootNode, PlanningSet planningSet, UserSession session, BitControl.QueryContextInformation queryContextInfo)
double	getAffinityFactor()
int	getMaxGlobalWidth()
int	getMaxWidthPerNode()
protected abstract BiFunction<CoordinationProtos.DrillbitEndpoint,PhysicalOperator,Long>	getMemory()
Set<Wrapper>	getRootFragments(PlanningSet planningSet)
long	getSliceTarget()
List<QueryWorkUnit>	getSplitFragments(OptionList options, CoordinationProtos.DrillbitEndpoint foremanNode, UserBitShared.QueryId queryId, Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootFragment, UserSession session, BitControl.QueryContextInformation queryContextInfo)	Create multiple physical plans from original query planning, it will allow execute them eventually independently
void	initFragmentWrappers(Fragment rootFragment, PlanningSet planningSet)
PlanningSet	prepareFragmentTree(Fragment rootFragment)
protected void	traverse(Wrapper fragmentWrapper, Consumer<Wrapper> operation)	Call operation on each fragment.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

SimpleParallelizer

protected SimpleParallelizer(QueryContext context)

SimpleParallelizer

protected SimpleParallelizer(long parallelizationThreshold,
 int maxWidthPerNode,
 int maxGlobalWidth,
 double affinityFactor)

Method Details

getSliceTarget

public long getSliceTarget()

Specified by:

getSliceTarget in interface ParallelizationParameters

Returns:

Configured max width per slice of work.

getMaxWidthPerNode

public int getMaxWidthPerNode()

Specified by:

getMaxWidthPerNode in interface ParallelizationParameters

Returns:

Configured maximum allowed number of parallelization units per node.

getMaxGlobalWidth

public int getMaxGlobalWidth()

Specified by:

getMaxGlobalWidth in interface ParallelizationParameters

Returns:

Configured maximum allowed number of parallelization units per all nodes in the cluster.

getAffinityFactor

public double getAffinityFactor()

Specified by:

getAffinityFactor in interface ParallelizationParameters

Returns:

Factor by which a node with endpoint affinity will be favored while creating assignment.

getRootFragments

public Set<Wrapper> getRootFragments(PlanningSet planningSet)

prepareFragmentTree

public PlanningSet prepareFragmentTree(Fragment rootFragment)

collectStatsAndParallelizeFragments

public void collectStatsAndParallelizeFragments(PlanningSet planningSet,
 Set<Wrapper> roots,
 Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints)
                                         throws PhysicalOperatorSetupException

Traverse all the major fragments and parallelize each major fragment based on collected stats. The children fragments are parallelized before a parent fragment.

Parameters:

planningSet - Set of all major fragments and their context.

roots - Root nodes of the plan.

activeEndpoints - currently active drillbit endpoints.

Throws:

PhysicalOperatorSetupException

adjustMemory

public abstract void adjustMemory(PlanningSet planningSet,
 Set<Wrapper> roots,
 Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints)
                           throws PhysicalOperatorSetupException

Throws:

PhysicalOperatorSetupException

generateWorkUnit

public final QueryWorkUnit generateWorkUnit(OptionList options,
 CoordinationProtos.DrillbitEndpoint foremanNode,
 UserBitShared.QueryId queryId,
 Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints,
 Fragment rootFragment,
 UserSession session,
 BitControl.QueryContextInformation queryContextInfo)
                                     throws ExecutionSetupException

The starting function for the whole parallelization and memory computation logic. 1) Initially a fragment tree is prepared which contains a wrapper for each fragment. The topology of this tree is same as that of the major fragment tree. 2) Traverse this fragment tree to collect stats for each major fragment and then parallelize each fragment. At this stage minor fragments are not created but all the required information to create minor fragment are computed. 3) Memory is computed for each operator and for the minor fragment. 4) Lastly all the above computed information is used to create the minor fragments for each major fragment.

Specified by:

generateWorkUnit in interface QueryParallelizer

Parameters:

options - List of options set by the user.

foremanNode - foreman node for this query plan.

queryId - Query ID.

activeEndpoints - currently active endpoints on which this plan will run.

rootFragment - Root major fragment.

session - session context.

queryContextInfo - query context.

Returns:

Throws:

ExecutionSetupException

getSplitFragments

public List<QueryWorkUnit> getSplitFragments(OptionList options,
 CoordinationProtos.DrillbitEndpoint foremanNode,
 UserBitShared.QueryId queryId,
 Collection<CoordinationProtos.DrillbitEndpoint> activeEndpoints,
 PhysicalPlanReader reader,
 Fragment rootFragment,
 UserSession session,
 BitControl.QueryContextInformation queryContextInfo)
                                      throws ExecutionSetupException

Create multiple physical plans from original query planning, it will allow execute them eventually independently

Parameters:

options -

foremanNode -

queryId -

activeEndpoints -

reader -

rootFragment -

session -

queryContextInfo -

Returns:

The QueryWorkUnits.

Throws:

ExecutionSetupException

initFragmentWrappers

public void initFragmentWrappers(Fragment rootFragment,
 PlanningSet planningSet)

traverse

protected void traverse(Wrapper fragmentWrapper,
 Consumer<Wrapper> operation)
                 throws PhysicalOperatorSetupException

Call operation on each fragment. Traversal calls operation on child fragments before calling it on the parent fragment.

Throws:

PhysicalOperatorSetupException

getMemory

protected abstract BiFunction<CoordinationProtos.DrillbitEndpoint,PhysicalOperator,Long> getMemory()

generateWorkUnit

protected QueryWorkUnit generateWorkUnit(OptionList options,
 CoordinationProtos.DrillbitEndpoint foremanNode,
 UserBitShared.QueryId queryId,
 Fragment rootNode,
 PlanningSet planningSet,
 UserSession session,
 BitControl.QueryContextInformation queryContextInfo)
                                  throws ExecutionSetupException

Throws:

ExecutionSetupException