You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2013/12/28 07:36:36 UTC
[39/50] [abbrv] Solving conflicts
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
----------------------------------------------------------------------
diff --cc tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
index 9190da7,ea3c366..0e37ea4
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
@@@ -23,9 -25,9 +25,11 @@@ import com.google.common.collect.Sets
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
++import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.algebra.JoinType;
import org.apache.tajo.catalog.*;
+ import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.eval.AggregationFunctionCallEval;
import org.apache.tajo.engine.eval.EvalTreeUtil;
@@@ -70,109 -68,44 +71,63 @@@ public class GlobalPlanner
globalPlanContext.plan = masterPlan;
LOG.info(masterPlan.getLogicalPlan());
- LogicalNode rootNode = PlannerUtil.clone(masterPlan.getLogicalPlan().getRootBlock().getRoot());
- planner.visitChild(globalPlanContext, masterPlan.getLogicalPlan(), rootNode, new Stack<LogicalNode>());
-
- ExecutionBlock terminalBlock = masterPlan.createTerminalBlock();
-
- if (globalPlanContext.lastRepartionableNode != null
- && globalPlanContext.lastRepartionableNode.getType() == NodeType.UNION) {
- UnionNode unionNode = (UnionNode) globalPlanContext.lastRepartionableNode;
- ConsecutiveUnionFinder finder = new ConsecutiveUnionFinder();
- UnionsFinderContext finderContext = new UnionsFinderContext();
- finder.visitChild(finderContext, masterPlan.getLogicalPlan(), unionNode, new Stack<LogicalNode>());
-
- for (UnionNode union : finderContext.unionList) {
- TableSubQueryNode leftSubQuery = union.getLeftChild();
- TableSubQueryNode rightSubQuery = union.getRightChild();
- if (leftSubQuery.getSubQuery().getType() != NodeType.UNION) {
- ExecutionBlock execBlock = masterPlan.newExecutionBlock();
- execBlock.setPlan(leftSubQuery);
- int childPid = execBlock.getPlan().getChild(execBlock.getPlan().getTerminalNode(), 0).getPID();
- DataChannel dataChannel = new DataChannel(execBlock, terminalBlock, childPid, null, NONE_PARTITION, 1,
- execBlock.getPlan().getOutSchema(0));
- masterPlan.addConnect(dataChannel);
- }
- if (rightSubQuery.getSubQuery().getType() != NodeType.UNION) {
- ExecutionBlock execBlock = masterPlan.newExecutionBlock();
- execBlock.setPlan(rightSubQuery);
- int childPid = execBlock.getPlan().getChild(execBlock.getPlan().getTerminalNode(), 0).getPID();
- DataChannel dataChannel = new DataChannel(execBlock, terminalBlock, childPid, null, NONE_PARTITION, 1,
- execBlock.getPlan().getOutSchema(0));
- masterPlan.addConnect(dataChannel);
- }
- }
- } else {
- int childPid = globalPlanContext.topMostLeftExecBlock.getPlan().getChild(
- globalPlanContext.topMostLeftExecBlock.getPlan().getTerminalNode(), 0).getPID();
- DataChannel dataChannel = new DataChannel(globalPlanContext.topMostLeftExecBlock, terminalBlock, childPid, null,
- NONE_PARTITION, 1, globalPlanContext.topMostLeftExecBlock.getPlan().getOutSchema(0));
- LogicalNode inputPlan = PlannerUtil.clone(masterPlan.getLogicalPlan(),
++ LogicalNode inputPlan = PlannerUtil.clone(masterPlan.getLogicalPlan().getPidFactory(),
+ masterPlan.getLogicalPlan().getRootBlock().getRoot());
+ LogicalNode lastNode = planner.visitChild(globalPlanContext, masterPlan.getLogicalPlan(), inputPlan,
+ new Stack<LogicalNode>());
+
+ ExecutionBlock childExecBlock = globalPlanContext.execBlockMap.get(lastNode.getPID());
+
+ if (childExecBlock.getPlan() != null) {
+ ExecutionBlock terminalBlock = masterPlan.createTerminalBlock();
- DataChannel dataChannel = new DataChannel(childExecBlock, terminalBlock, NONE_PARTITION, 1);
++ DataChannel dataChannel = new DataChannel(childExecBlock, terminalBlock, lastNode.getPID(), -1, NONE_PARTITION, 1,
++ lastNode.getOutSchema());
+ dataChannel.setStoreType(CatalogProtos.StoreType.CSV);
- dataChannel.setSchema(lastNode.getOutSchema());
masterPlan.addConnect(dataChannel);
+ masterPlan.setTerminal(terminalBlock);
+ } else {
+ masterPlan.setTerminal(childExecBlock);
}
- masterPlan.setTerminal(terminalBlock);
+
LOG.info(masterPlan);
}
- private ExecutionBlock buildRepartitionBlocks(MasterPlan masterPlan, LogicalNode lastDistNode, LogicalNode curNode,
- LogicalNode childNode, ExecutionBlock lastChildBlock)
- throws PlanningException {
-
- ExecutionBlock currentBlock = null;
- ExecutionBlock childBlock = lastChildBlock;
-
- NodeType shuffleRequiredNodeType = lastDistNode.getType();
- if (shuffleRequiredNodeType == NodeType.GROUP_BY) {
- ExecutionBlock [] blocks = buildGroupBy(masterPlan, lastDistNode, curNode, childNode, childBlock);
- currentBlock = blocks[0];
- } else if (shuffleRequiredNodeType == NodeType.SORT) {
- ExecutionBlock [] blocks = buildSortPlan(masterPlan, lastDistNode, curNode, childNode, childBlock);
- currentBlock = blocks[0];
- } else if (shuffleRequiredNodeType == NodeType.JOIN) {
- ExecutionBlock [] blocks = buildJoinPlan(masterPlan, lastDistNode, childBlock, lastChildBlock);
- currentBlock = blocks[0];
- }
-
- return currentBlock;
++ public static ScanNode buildInputExecutor(LogicalPlan plan, Schema schema, ExecutionBlockId srcId, ExecutionBlockId targetId, StoreType storeType) {
++ Preconditions.checkArgument(schema != null,
++ "Channel schema (" + srcId +" -> "+ targetId + ") is not initialized");
++ TableMeta meta = new TableMeta(storeType, new Options());
++ TableDesc desc = new TableDesc(srcId.toString(), schema, meta, new Path("/"));
++ return new ScanNode(plan.newPID(), desc);
+ }
+
public static ScanNode buildInputExecutor(LogicalPlan plan, DataChannel channel) {
Preconditions.checkArgument(channel.getSchema() != null,
- "Channel schema (" + channel.getSrcId().getId() + " -> " + channel.getTargetId().getId() + ") is not initialized");
+ "Channel schema (" + channel.getSrcId().getId() +" -> "+ channel.getTargetId().getId()+") is not initialized");
TableMeta meta = new TableMeta(channel.getStoreType(), new Options());
TableDesc desc = new TableDesc(channel.getSrcId().toString(), channel.getSchema(), meta, new Path("/"));
return new ScanNode(plan.newPID(), desc);
}
- public static ScanNode buildInputExecutor(LogicalPlan plan, String tableName, Schema schema, StoreType storeType) {
- TableMeta meta = new TableMeta(storeType, new Options());
- TableDesc desc = new TableDesc(tableName, schema, meta, new Path("/"));
- return new ScanNode(plan.newPID(), desc);
- }
-
- private static String getIntermediateTableName(ExecutionBlock block) {
- return block.getId().toString();
- }
-
private DataChannel createDataChannelFromJoin(ExecutionBlock leftBlock, ExecutionBlock rightBlock,
ExecutionBlock parent, JoinNode join, boolean leftTable) {
- ExecutionBlock childBlock = leftTable ? leftBlock : rightBlock;
+ ExecutionBlock childBlock;
- int childPid, parentPid;
- ExecutionPlan execPlan = parent.getPlan();
- LogicalNode scanParent = execPlan.getParent(execPlan.getInputContext().getScanNodes()[0], 0);
++ int targetPid;
+ if (leftTable) {
+ childBlock = leftBlock;
- parentPid = execPlan.getChild(scanParent, Tag.LEFT).getPID();
++ targetPid = parent.getInputContext().getScanNodes()[0].getPID();
+ } else {
+ childBlock = rightBlock;
- parentPid = execPlan.getChild(scanParent, Tag.RIGHT).getPID();
++ targetPid = parent.getInputContext().getScanNodes()[1].getPID();
+ }
- childPid = childBlock.getPlan().getChild(childBlock.getPlan().getTerminalNode(), 0).getPID();
++ LogicalNode srcTopNode = childBlock.getPlan().getChild(childBlock.getPlan().getTerminalNode(), 0);
++ int srcPid = srcTopNode.getPID();
- DataChannel channel = new DataChannel(childBlock, parent, childPid, parentPid, HASH_PARTITION, 32,
- childBlock.getPlan().getOutSchema(0));
- DataChannel channel = new DataChannel(childBlock, parent, HASH_PARTITION, 32);
++ DataChannel channel = new DataChannel(childBlock, parent, srcPid, targetPid, HASH_PARTITION, 32,
++ srcTopNode.getOutSchema());
+ channel.setStoreType(storeType);
if (join.getJoinType() != JoinType.CROSS) {
Column [][] joinColumns = PlannerUtil.joinJoinKeyForEachTable(join.getJoinQual(),
- join.getLeftChild().getOutSchema(), join.getRightChild().getOutSchema());
- leftBlock.getPlan().getOutSchema(), rightBlock.getPlan().getOutSchema());
++ leftBlock.getPlan().getOutSchema(0), rightBlock.getPlan().getOutSchema(0));
if (leftTable) {
channel.setPartitionKey(joinColumns[0]);
} else {
@@@ -248,143 -174,114 +196,118 @@@
masterPlan.addConnect(leftChannel);
masterPlan.addConnect(rightChannel);
- return new ExecutionBlock[] { currentBlock, childBlock };
+ return currentBlock;
}
- private ExecutionBlock [] buildGroupBy(MasterPlan masterPlan, LogicalNode lastDistNode, LogicalNode currentNode,
- LogicalNode childNode, ExecutionBlock childBlock) throws PlanningException {
- ExecutionBlock currentBlock = null;
- GroupbyNode groupByNode = (GroupbyNode) lastDistNode;
-
- if (groupByNode.isDistinct()) {
- if (childBlock == null) { // first repartition node
- childBlock = masterPlan.newExecutionBlock();
- }
- childBlock.setPlan(groupByNode.getChild());
- currentBlock = masterPlan.newExecutionBlock();
-
- LinkedHashSet<Column> columnsForDistinct = new LinkedHashSet<Column>();
-
- for (Target target : groupByNode.getTargets()) {
- List<AggregationFunctionCallEval> functions = EvalTreeUtil.findDistinctAggFunction(target.getEvalTree());
- for (AggregationFunctionCallEval function : functions) {
- if (function.isDistinct()) {
- columnsForDistinct.addAll(EvalTreeUtil.findDistinctRefColumns(function));
- }
+ /**
+ * If a query contains a distinct aggregation function, the query does not
+ * perform pre-aggregation in the first phase. Instead, in the fist phase,
+ * the query performs only hash shuffle. Then, the query performs the
+ * sort aggregation in the second phase. At that time, the aggregation
+ * function should be executed as the first phase.
+ */
+ private ExecutionBlock buildDistinctGroupBy(GlobalPlanContext context, ExecutionBlock childBlock,
+ GroupbyNode groupbyNode) {
+ // setup child block
+ LogicalNode topMostOfFirstPhase = groupbyNode.getChild();
+ childBlock.setPlan(topMostOfFirstPhase);
+
+ // setup current block
+ ExecutionBlock currentBlock = context.plan.newExecutionBlock();
+ LinkedHashSet<Column> columnsForDistinct = new LinkedHashSet<Column>();
+ for (Target target : groupbyNode.getTargets()) {
+ List<AggregationFunctionCallEval> functions = EvalTreeUtil.findDistinctAggFunction(target.getEvalTree());
+ for (AggregationFunctionCallEval function : functions) {
+ if (function.isDistinct()) {
+ columnsForDistinct.addAll(EvalTreeUtil.findDistinctRefColumns(function));
+ } else {
+ // See the comment of this method. the aggregation function should be executed as the first phase.
+ function.setFirstPhase();
}
}
+ }
- Set<Column> existingColumns = Sets.newHashSet(groupByNode.getGroupingColumns());
- columnsForDistinct.removeAll(existingColumns); // remove existing grouping columns
- SortSpec [] sortSpecs = PlannerUtil.columnsToSortSpec(columnsForDistinct);
- currentBlock.getEnforcer().enforceSortAggregation(groupByNode.getPID(), sortSpecs);
-
- GroupbyNode secondGroupBy = groupByNode;
- ScanNode scanNode = buildInputExecutor(masterPlan.getLogicalPlan(), getIntermediateTableName(childBlock),
- childBlock.getPlan().getOutSchema(0), StoreType.CSV);
- secondGroupBy.setChild(scanNode);
-
- LogicalNode parent = PlannerUtil.findTopParentNode(currentNode, lastDistNode.getType());
- if (parent instanceof UnaryNode && parent != secondGroupBy) {
- ((UnaryNode)parent).setChild(secondGroupBy);
- }
-
- currentBlock.setPlan(currentNode);
+ // Set sort aggregation enforcer to the second groupby node
+ Set<Column> existingColumns = Sets.newHashSet(groupbyNode.getGroupingColumns());
+ columnsForDistinct.removeAll(existingColumns); // remove existing grouping columns
+ SortSpec [] sortSpecs = PlannerUtil.columnsToSortSpec(columnsForDistinct);
+ currentBlock.getEnforcer().enforceSortAggregation(groupbyNode.getPID(), sortSpecs);
- int childPid = childBlock.getPlan().getChild(childBlock.getPlan().getTerminalNode(), 0).getPID();
- int parentPid = currentBlock.getInputContext().getScanNodes()[0].getPID();
- DataChannel channel = new DataChannel(childBlock, currentBlock, childPid, parentPid, HASH_PARTITION, 32,
- childBlock.getPlan().getOutSchema(0));
- channel.setPartitionKey(groupByNode.getGroupingColumns());
- masterPlan.addConnect(channel);
++ // setup current block with channel
++ ScanNode scanNode = buildInputExecutor(context.plan.getLogicalPlan(), topMostOfFirstPhase.getOutSchema(),
++ childBlock.getId(), currentBlock.getId(), storeType);
++ groupbyNode.setChild(scanNode);
++ currentBlock.setPlan(groupbyNode);
- } else {
+ // setup channel
+ DataChannel channel;
- channel = new DataChannel(childBlock, currentBlock, HASH_PARTITION, 32);
++ channel = new DataChannel(childBlock, currentBlock, topMostOfFirstPhase.getPID(), scanNode.getPID(), HASH_PARTITION,
++ 32, topMostOfFirstPhase.getOutSchema());
+ channel.setPartitionKey(groupbyNode.getGroupingColumns());
- channel.setSchema(topMostOfFirstPhase.getOutSchema());
+ channel.setStoreType(storeType);
- GroupbyNode[] twoPhaseGroupby = PlannerUtil.transformGroupbyTo2Pv2(masterPlan.getLogicalPlan(), groupByNode);
- GroupbyNode secondGroupBy = twoPhaseGroupby[0];
- GroupbyNode firstGroupBy = twoPhaseGroupby[1];
- firstGroupBy.setHavingCondition(null);
- // setup current block with channel
- ScanNode scanNode = buildInputExecutor(context.plan.getLogicalPlan(), channel);
- groupbyNode.setChild(scanNode);
- currentBlock.setPlan(groupbyNode);
+ context.plan.addConnect(channel);
- int childPid, parentPid;
- if (firstGroupBy.getChild().getType() == NodeType.TABLE_SUBQUERY &&
- ((TableSubQueryNode)firstGroupBy.getChild()).getSubQuery().getType() == NodeType.UNION) {
+ return currentBlock;
+ }
- UnionNode unionNode = PlannerUtil.findTopNode(firstGroupBy, NodeType.UNION);
- ConsecutiveUnionFinder finder = new ConsecutiveUnionFinder();
- UnionsFinderContext finderContext = new UnionsFinderContext();
- finder.visitChild(finderContext, masterPlan.getLogicalPlan(), unionNode, new Stack<LogicalNode>());
+ private ExecutionBlock buildGroupBy(GlobalPlanContext context, ExecutionBlock childBlock,
+ GroupbyNode groupbyNode)
+ throws PlanningException {
- currentBlock = masterPlan.newExecutionBlock();
- LogicalNode parent = PlannerUtil.findTopParentNode(currentNode, lastDistNode.getType());
- if (parent instanceof UnaryNode && parent != secondGroupBy) {
- ((UnaryNode)parent).setChild(secondGroupBy);
- }
+ MasterPlan masterPlan = context.plan;
+ ExecutionBlock currentBlock;
- for (UnionNode union : finderContext.unionList) {
- TableSubQueryNode leftSubQuery = union.getLeftChild();
- TableSubQueryNode rightSubQuery = union.getRightChild();
- DataChannel dataChannel;
- if (leftSubQuery.getSubQuery().getType() != NodeType.UNION) {
- ExecutionBlock execBlock = masterPlan.newExecutionBlock();
- GroupbyNode g1 = PlannerUtil.clone(firstGroupBy);
- g1.setChild(leftSubQuery);
- execBlock.setPlan(g1);
- ScanNode scanNode = buildInputExecutor(masterPlan.getLogicalPlan(),
- getIntermediateTableName(execBlock), execBlock.getPlan().getOutSchema(0), StoreType.CSV);
- secondGroupBy.setChild(scanNode);
- currentBlock.setPlan(currentNode);
- childPid = execBlock.getPlan().getChild(execBlock.getPlan().getTerminalNode(), 0).getPID();
- parentPid = currentBlock.getInputContext().getScanNodes()[0].getPID();
- dataChannel = new DataChannel(execBlock, currentBlock, childPid, parentPid, HASH_PARTITION, 32,
- execBlock.getPlan().getOutSchema(0));
- masterPlan.addConnect(dataChannel);
- }
- if (rightSubQuery.getSubQuery().getType() != NodeType.UNION) {
- ExecutionBlock execBlock = masterPlan.newExecutionBlock();
- GroupbyNode g1 = PlannerUtil.clone(firstGroupBy);
- g1.setChild(rightSubQuery);
- execBlock.setPlan(g1);
- ScanNode scanNode = buildInputExecutor(masterPlan.getLogicalPlan(),
- getIntermediateTableName(execBlock), execBlock.getPlan().getOutSchema(0), StoreType.CSV);
- secondGroupBy.setChild(scanNode);
- currentBlock.setPlan(currentNode);
- childPid = execBlock.getPlan().getChild(execBlock.getPlan().getTerminalNode(), 0).getPID();
- parentPid = currentBlock.getInputContext().getScanNodes()[0].getPID();
- dataChannel = new DataChannel(execBlock, currentBlock, childPid, parentPid, HASH_PARTITION, 32,
- execBlock.getPlan().getOutSchema(0));
- masterPlan.addConnect(dataChannel);
+ if (groupbyNode.isDistinct()) {
+ return buildDistinctGroupBy(context, childBlock, groupbyNode);
+ } else {
+ GroupbyNode firstPhaseGroupBy = PlannerUtil.transformGroupbyTo2P(groupbyNode);
+ firstPhaseGroupBy.setHavingCondition(null);
+
+ if (firstPhaseGroupBy.getChild().getType() == NodeType.TABLE_SUBQUERY &&
+ ((TableSubQueryNode)firstPhaseGroupBy.getChild()).getSubQuery().getType() == NodeType.UNION) {
+
+ currentBlock = childBlock;
+ for (DataChannel dataChannel : masterPlan.getIncomingChannels(currentBlock.getId())) {
+ if (firstPhaseGroupBy.isEmptyGrouping()) {
+ dataChannel.setPartition(HASH_PARTITION, firstPhaseGroupBy.getGroupingColumns(), 1);
+ } else {
+ dataChannel.setPartition(HASH_PARTITION, firstPhaseGroupBy.getGroupingColumns(), 32);
}
+ dataChannel.setSchema(firstPhaseGroupBy.getOutSchema());
+
+ ExecutionBlock subBlock = masterPlan.getExecBlock(dataChannel.getSrcId());
- GroupbyNode g1 = PlannerUtil.clone(context.plan.getLogicalPlan(), firstPhaseGroupBy);
- g1.setChild(subBlock.getPlan());
++ ExecutionPlan subBlockPlan = subBlock.getPlan();
++ LogicalNode topNode = subBlockPlan.getChild(subBlockPlan.getTerminalNode(), 0);
++ GroupbyNode g1 = PlannerUtil.clone(context.plan.getLogicalPlan().getPidFactory(), firstPhaseGroupBy);
++ subBlockPlan.add(topNode, g1, Tag.SINGLE);
++// g1.setChild(subBlock.getPlan());
+ subBlock.setPlan(g1);
+
- GroupbyNode g2 = PlannerUtil.clone(context.plan.getLogicalPlan(), groupbyNode);
++ GroupbyNode g2 = PlannerUtil.clone(context.plan.getLogicalPlan().getPidFactory(), groupbyNode);
+ ScanNode scanNode = buildInputExecutor(masterPlan.getLogicalPlan(), dataChannel);
+ g2.setChild(scanNode);
+ currentBlock.setPlan(g2);
}
- } else {
- if (childBlock == null) { // first repartition node
- childBlock = masterPlan.newExecutionBlock();
- }
- childBlock.setPlan(firstGroupBy);
-
- LogicalNode parent = PlannerUtil.findTopParentNode(currentNode, lastDistNode.getType());
- if (parent instanceof UnaryNode && parent != secondGroupBy) {
- ((UnaryNode)parent).setChild(secondGroupBy);
- }
-
+ } else { // general hash-shuffled aggregation
+ childBlock.setPlan(firstPhaseGroupBy);
currentBlock = masterPlan.newExecutionBlock();
- ScanNode scanNode = buildInputExecutor(masterPlan.getLogicalPlan(),
- getIntermediateTableName(childBlock), childBlock.getPlan().getOutSchema(0), StoreType.CSV);
- secondGroupBy.setChild(scanNode);
- currentBlock.setPlan(currentNode);
DataChannel channel;
- childPid = childBlock.getPlan().getChild(childBlock.getPlan().getTerminalNode(), 0).getPID();
- parentPid = currentBlock.getInputContext().getScanNodes()[0].getPID();
- if (firstGroupBy.isEmptyGrouping()) {
- channel = new DataChannel(childBlock, currentBlock, childPid, parentPid, HASH_PARTITION, 1,
- childBlock.getPlan().getOutSchema(0));
- channel.setPartitionKey(firstGroupBy.getGroupingColumns());
+ if (firstPhaseGroupBy.isEmptyGrouping()) {
+ channel = new DataChannel(childBlock, currentBlock, HASH_PARTITION, 1);
+ channel.setPartitionKey(firstPhaseGroupBy.getGroupingColumns());
} else {
- channel = new DataChannel(childBlock, currentBlock, childPid, parentPid, HASH_PARTITION, 32,
- childBlock.getPlan().getOutSchema(0));
- channel.setPartitionKey(firstGroupBy.getGroupingColumns());
+ channel = new DataChannel(childBlock, currentBlock, HASH_PARTITION, 32);
+ channel.setPartitionKey(firstPhaseGroupBy.getGroupingColumns());
}
+ channel.setSchema(firstPhaseGroupBy.getOutSchema());
+ channel.setStoreType(storeType);
+ ScanNode scanNode = buildInputExecutor(masterPlan.getLogicalPlan(), channel);
+ groupbyNode.setChild(scanNode);
+ groupbyNode.setInSchema(scanNode.getOutSchema());
+ currentBlock.setPlan(groupbyNode);
masterPlan.addConnect(channel);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/global/MasterPlan.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/LimitNode.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/LogicalNode.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/ScanNode.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PartitionedStoreExec.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
----------------------------------------------------------------------
diff --cc tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
index a214adb,b1deb43..6af605d
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
@@@ -467,12 -469,10 +469,12 @@@ public class DefaultTaskScheduler exten
new ArrayList<FragmentProto>(task.getAllFragments()),
"",
false,
- task.getLogicalPlan().toJson(),
+ task.getPlan().toJson(),
context.getQueryContext(),
- subQuery.getDataChannel(), subQuery.getBlock().getEnforcer());
+ subQuery.getIncomingChannels(),
+ subQuery.getOutgoingChannels(),
+ subQuery.getBlock().getEnforcer());
- if (!subQuery.getMasterPlan().isRoot(subQuery.getBlock())) {
+ if (checkIfInterQuery(subQuery.getMasterPlan(), subQuery.getBlock())) {
taskAssign.setInterQuery();
}
@@@ -517,12 -530,11 +532,12 @@@
Lists.newArrayList(task.getAllFragments()),
"",
false,
- task.getLogicalPlan().toJson(),
+ task.getPlan().toJson(),
context.getQueryContext(),
- subQuery.getDataChannel(),
+ subQuery.getIncomingChannels(),
+ subQuery.getOutgoingChannels(),
subQuery.getBlock().getEnforcer());
- if (!subQuery.getMasterPlan().isRoot(subQuery.getBlock())) {
+ if (checkIfInterQuery(subQuery.getMasterPlan(), subQuery.getBlock())) {
taskAssign.setInterQuery();
}
for (ScanNode scan : task.getScanNodes()) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
----------------------------------------------------------------------
diff --cc tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
index 29e6387,03659c3..894da09
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
@@@ -595,10 -692,11 +697,11 @@@ public class SubQuery implements EventH
return maxTaskNum;
}
- public static long getInputVolume(MasterPlan masterPlan, QueryMasterTask.QueryMasterTaskContext context, ExecutionBlock execBlock) {
+ public static long getInputVolume(MasterPlan masterPlan, QueryMasterTask.QueryMasterTaskContext context,
+ ExecutionBlock execBlock) {
Map<String, TableDesc> tableMap = context.getTableDescMap();
if (masterPlan.isLeaf(execBlock)) {
- ScanNode outerScan = execBlock.getScanNodes()[0];
+ ScanNode outerScan = execBlock.getInputContext().getScanNodes()[0];
TableStats stat = tableMap.get(outerScan.getCanonicalName()).getStats();
return stat.getNumBytes();
} else {
@@@ -640,20 -742,42 +747,41 @@@
subQuery.eventHandler.handle(event);
}
+ /**
+ * It creates a number of fragments for all partitions.
+ */
+ private static Collection<FileFragment> getFragmentsFromPartitionedTable(SubQuery subQuery,
+ ScanNode scan,
+ TableDesc table) throws IOException {
+ List<FileFragment> fragments = Lists.newArrayList();
+ PartitionedTableScanNode partitionsScan = (PartitionedTableScanNode) scan;
+ for (Path path : partitionsScan.getInputPaths()) {
+ fragments.addAll(subQuery.getStorageManager().getSplits(
+ scan.getCanonicalName(), table.getMeta(), table.getSchema(), path));
+ }
+ partitionsScan.setInputPaths(null);
+ return fragments;
+ }
+
private static QueryUnit [] createLeafTasks(SubQuery subQuery) throws IOException {
ExecutionBlock execBlock = subQuery.getBlock();
- TableMeta meta;
- Path inputPath;
-
- ScanNode[] scans = execBlock.getScanNodes();
- Preconditions.checkArgument(scans.length == 1, "Must be Scan Query");
- ScanNode scan = scans[0];
+ Preconditions.checkState(execBlock.getInputContext().size() == 1);
- ScanNode scanNode = execBlock.getInputContext().getScanNodes()[0];
- TableDesc desc = subQuery.context.getTableDescMap().get(scanNode.getCanonicalName());
- inputPath = desc.getPath();
- meta = desc.getMeta();
-
- // TODO - should be change the inner directory
- List<FileFragment> fragments = subQuery.getStorageManager().getSplits(scanNode.getCanonicalName(),
- meta, desc.getSchema(), inputPath);
++ ScanNode scan = execBlock.getInputContext().getScanNodes()[0];
+ TableDesc table = subQuery.context.getTableDescMap().get(scan.getCanonicalName());
+
+ Collection<FileFragment> fragments;
+ TableMeta meta = table.getMeta();
+
+ // Depending on scanner node's type, it creates fragments. If scan is for
+ // a partitioned table, It will creates lots fragments for all partitions.
+ // Otherwise, it creates at least one fragments for a table, which may
+ // span a number of blocks or possibly consists of a number of files.
+ if (scan.getType() == NodeType.PARTITIONS_SCAN) {
+ fragments = getFragmentsFromPartitionedTable(subQuery, scan, table);
+ } else {
+ Path inputPath = table.getPath();
+ fragments = subQuery.getStorageManager().getSplits(scan.getCanonicalName(), meta, table.getSchema(), inputPath);
+ }
QueryUnit queryUnit;
List<QueryUnit> queryUnits = new ArrayList<QueryUnit>();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalPlan.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestHashAntiJoinExec.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestHashSemiJoinExec.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
----------------------------------------------------------------------
diff --cc tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
index 76d54e0,1a46af6..078d24d
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
@@@ -472,14 -450,12 +472,14 @@@ public class TestPhysicalPlanner
Column key1 = new Column("score.deptName", Type.TEXT);
Column key2 = new Column("score.class", Type.TEXT);
DataChannel dataChannel = new DataChannel(masterPlan.newExecutionBlockId(), masterPlan.newExecutionBlockId(),
- PartitionType.HASH_PARTITION, numPartitions);
+ ((LogicalRootNode)rootNode).getChild().getPID(), null, PartitionType.HASH_PARTITION, numPartitions);
+ dataChannel.setSchema(rootNode.getOutSchema());
dataChannel.setPartitionKey(new Column[]{key1, key2});
- ctx.setDataChannel(dataChannel);
- LogicalNode rootNode = optimizer.optimize(plan);
+ List<DataChannel> channels = new ArrayList<DataChannel>();
+ channels.add(dataChannel);
+ ctx.setOutgoingChannels(channels);
- TableMeta outputMeta = CatalogUtil.newTableMeta(StoreType.CSV);
+ TableMeta outputMeta = CatalogUtil.newTableMeta(dataChannel.getStoreType());
FileSystem fs = sm.getFileSystem();
@@@ -531,24 -505,20 +531,24 @@@
ctx.setEnforcer(new Enforcer());
Expr expr = analyzer.parse(QUERIES[14]);
LogicalPlan plan = planner.createPlan(expr);
- LogicalNode rootNode = plan.getRootBlock().getRoot();
+ LogicalNode rootNode = optimizer.optimize(plan);
int numPartitions = 1;
DataChannel dataChannel = new DataChannel(masterPlan.newExecutionBlockId(), masterPlan.newExecutionBlockId(),
- PartitionType.HASH_PARTITION, numPartitions);
+ ((LogicalRootNode)rootNode).getChild().getPID(), null, PartitionType.HASH_PARTITION, numPartitions);
+ dataChannel.setSchema(rootNode.getOutSchema());
dataChannel.setPartitionKey(new Column[]{});
- ctx.setDataChannel(dataChannel);
- optimizer.optimize(plan);
+ List<DataChannel> channels = new ArrayList<DataChannel>();
+ channels.add(dataChannel);
+ ctx.setOutgoingChannels(channels);
- TableMeta outputMeta = CatalogUtil.newTableMeta(StoreType.CSV);
+ TableMeta outputMeta = CatalogUtil.newTableMeta(dataChannel.getStoreType());
- PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf,sm);
- PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
+ ExecutionPlan execPlan = new ExecutionPlan();
+ execPlan.addPlan(rootNode);
+ PhysicalPlannerImpl phyPlanner = new PhysicalPlannerImpl(conf,sm);
+ PhysicalExec exec = phyPlanner.createPlan(ctx, execPlan);
exec.init();
- exec.next();
+ while (exec.next() != null);
exec.close();
Path path = new Path(workDir, "output");
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestSortExec.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/2495601c/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
----------------------------------------------------------------------
diff --cc tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
index 39408a0,2d97e7a..dd6ff01
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
@@@ -34,9 -34,11 +34,12 @@@ import org.apache.tajo.datum.Datum
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.*;
+import org.apache.tajo.engine.planner.global.ExecutionPlan;
import org.apache.tajo.engine.planner.logical.LogicalNode;
- import org.apache.tajo.engine.planner.physical.*;
+ import org.apache.tajo.engine.planner.physical.IndexedStoreExec;
+ import org.apache.tajo.engine.planner.physical.MemSortExec;
+ import org.apache.tajo.engine.planner.physical.PhysicalExec;
+ import org.apache.tajo.engine.planner.physical.ProjectionExec;
import org.apache.tajo.storage.*;
import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.storage.index.bst.BSTIndex;