You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2014/02/01 16:32:45 UTC
git commit: TAJO-574: Add a sort-based physical executor for column
partition store.
Updated Branches:
refs/heads/master d516fc499 -> 10c599f4b
TAJO-574: Add a sort-based physical executor for column partition store.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/10c599f4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/10c599f4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/10c599f4
Branch: refs/heads/master
Commit: 10c599f4b057308eca7ac8d5d7cc2542a69f0524
Parents: d516fc4
Author: Hyunsik Choi <hy...@apache.org>
Authored: Sat Feb 1 13:01:08 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Sun Feb 2 00:27:24 2014 +0900
----------------------------------------------------------------------
CHANGES.txt | 3 +
.../tajo/catalog/statistics/StatisticsUtil.java | 43 +++++
.../engine/planner/PhysicalPlannerImpl.java | 73 ++++++-
.../tajo/engine/planner/enforce/Enforcer.java | 12 ++
.../physical/BasicPhysicalExecutorVisitor.java | 71 ++++---
.../planner/physical/ColPartitionStoreExec.java | 100 ++++++++++
.../ColumnPartitionedTableStoreExec.java | 192 -------------------
.../HashBasedColPartitionStoreExec.java | 132 +++++++++++++
.../physical/PhysicalExecutorVisitor.java | 11 +-
.../SortBasedColPartitionStoreExec.java | 147 ++++++++++++++
.../src/main/proto/TajoWorkerProtocol.proto | 12 ++
.../planner/physical/TestPhysicalPlanner.java | 70 ++++++-
12 files changed, 634 insertions(+), 232 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index f038f97..10d002d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,9 @@ Release 0.8.0 - unreleased
NEW FEATURES
+ TAJO-574: Add a sort-based physical executor for column partition store.
+ (hyunsik)
+
TAJO-569: Add max(TEXT) function. (hyunsik)
TAJO-449: Implement extract() function. (Keuntae Park)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatisticsUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatisticsUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatisticsUtil.java
index b836818..8593db6 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatisticsUtil.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/StatisticsUtil.java
@@ -42,6 +42,49 @@ public class StatisticsUtil {
return aggregated;
}
+ /**
+ * Aggregate one table stats and accumulated stats, and then store it to the result stats.
+ *
+ * @param result It stores the aggregated stats
+ * @param stats The TableStats to be aggregated
+ */
+ public static void aggregateTableStat(TableStats result, TableStats stats) {
+
+ if (stats.getColumnStats().size() > 0) {
+ if (result.getColumnStats().size() == 0) {
+ for (int i = 0; i < stats.getColumnStats().size(); i++) {
+ result.addColumnStat(stats.getColumnStats().get(i));
+ }
+ } else {
+ // aggregate column stats for each table
+ for (int i = 0; i < stats.getColumnStats().size(); i++) {
+ ColumnStats cs = stats.getColumnStats().get(i);
+ ColumnStats agg = result.getColumnStats().get(i);
+ if (cs == null) {
+ LOG.warn("ERROR: One of column stats is NULL (expected column: " + agg.getColumn() + ")");
+ continue;
+ }
+
+ agg.setNumDistVals(agg.getNumDistValues() + cs.getNumDistValues());
+ agg.setNumNulls(agg.getNumNulls() + cs.getNumNulls());
+ if (!cs.minIsNotSet() && (agg.minIsNotSet() ||
+ agg.getMinValue().compareTo(cs.getMinValue()) > 0)) {
+ agg.setMinValue(cs.getMinValue());
+ }
+ if (!cs.maxIsNotSet() && (agg.maxIsNotSet() ||
+ agg.getMaxValue().compareTo(cs.getMaxValue()) < 0)) {
+ agg.setMaxValue(stats.getColumnStats().get(i).getMaxValue());
+ }
+ }
+ }
+ }
+
+ result.setNumRows(result.getNumRows() + stats.getNumRows());
+ result.setNumBytes(result.getNumBytes() + stats.getNumBytes());
+ result.setNumBlocks(result.getNumBlocks() + stats.getNumBlocks());
+ result.setNumShuffleOutputs(result.getNumShuffleOutputs() + stats.getNumShuffleOutputs());
+ }
+
public static TableStats aggregateTableStat(List<TableStats> tableStatses) {
TableStats aggregated = new TableStats();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index 202c59d..671cec3 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -47,6 +47,8 @@ import java.io.IOException;
import java.util.List;
import static org.apache.tajo.catalog.proto.CatalogProtos.FragmentProto;
+import static org.apache.tajo.catalog.proto.CatalogProtos.PartitionType;
+import static org.apache.tajo.ipc.TajoWorkerProtocol.ColumnPartitionEnforcer.ColumnPartitionAlgorithm;
import static org.apache.tajo.ipc.TajoWorkerProtocol.EnforceProperty;
import static org.apache.tajo.ipc.TajoWorkerProtocol.EnforceProperty.EnforceType;
import static org.apache.tajo.ipc.TajoWorkerProtocol.GroupbyEnforce.GroupbyAlgorithm;
@@ -75,7 +77,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
if (execPlan instanceof StoreTableExec
|| execPlan instanceof RangeShuffleFileWriteExec
|| execPlan instanceof HashShuffleFileWriteExec
- || execPlan instanceof ColumnPartitionedTableStoreExec) {
+ || execPlan instanceof ColPartitionStoreExec) {
return execPlan;
} else if (context.getDataChannel() != null) {
return buildOutputOperator(context, logicalPlan, execPlan);
@@ -677,7 +679,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
if (plan.getPartitionMethod() != null) {
switch (plan.getPartitionMethod().getPartitionType()) {
case COLUMN:
- return new ColumnPartitionedTableStoreExec(ctx, plan, subOp);
+ return createColumnPartitionStorePlan(ctx, plan, subOp);
default:
throw new IllegalStateException(plan.getPartitionMethod().getPartitionType() + " is not supported yet.");
}
@@ -686,6 +688,63 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
}
}
+ private PhysicalExec createColumnPartitionStorePlan(TaskAttemptContext context,
+ StoreTableNode storeTableNode,
+ PhysicalExec child) throws IOException {
+ Enforcer enforcer = context.getEnforcer();
+ EnforceProperty property = getAlgorithmEnforceProperty(enforcer, storeTableNode);
+ if (property != null) {
+ ColumnPartitionAlgorithm algorithm = property.getColumnPartition().getAlgorithm();
+ switch (algorithm) {
+ case HASH_PARTITION:
+ return createHashColumnPartitionStorePlan(context, storeTableNode, child);
+ case SORT_PARTITION: // default algorithm
+ default:
+ return createSortBasedColumnPartitionStorePlan(context, storeTableNode, child);
+ }
+ } else { // default algorithm is sorted-based column partition
+ return createSortBasedColumnPartitionStorePlan(context, storeTableNode, child);
+ }
+ }
+
+ private PhysicalExec createHashColumnPartitionStorePlan(TaskAttemptContext context,
+ StoreTableNode storeTableNode,
+ PhysicalExec child) throws IOException {
+ LOG.info("The planner chooses [Hash-based Column Partitioned Store] algorithm");
+ return new HashBasedColPartitionStoreExec(context, storeTableNode, child);
+ }
+
+ private PhysicalExec createSortBasedColumnPartitionStorePlan(TaskAttemptContext context,
+ StoreTableNode storeTableNode,
+ PhysicalExec child) throws IOException {
+
+ Column[] partitionKeyColumns = storeTableNode.getPartitionMethod().getExpressionSchema().toArray();
+ SortSpec[] sortSpecs = new SortSpec[partitionKeyColumns.length];
+
+ if (storeTableNode.getType() == NodeType.INSERT) {
+ InsertNode insertNode = (InsertNode) storeTableNode;
+ for (int i = 0; i < partitionKeyColumns.length; i++) {
+ for (Column column : partitionKeyColumns) {
+ int id = insertNode.getTableSchema().getColumnId(column.getQualifiedName());
+ sortSpecs[i++] = new SortSpec(insertNode.getProjectedSchema().getColumn(id), true, false);
+ }
+ }
+ } else {
+ for (int i = 0; i < partitionKeyColumns.length; i++) {
+ sortSpecs[i] = new SortSpec(partitionKeyColumns[i], true, false);
+ }
+ }
+
+ SortNode sortNode = new SortNode(-1);
+ sortNode.setSortSpecs(sortSpecs);
+ sortNode.setInSchema(child.getSchema());
+ sortNode.setOutSchema(child.getSchema());
+
+ ExternalSortExec sortExec = new ExternalSortExec(context, sm, sortNode, child);
+ LOG.info("The planner chooses [Sort-based Column Partitioned Store] algorithm");
+ return new SortBasedColPartitionStoreExec(context, storeTableNode, sortExec);
+ }
+
public PhysicalExec createScanPlan(TaskAttemptContext ctx, ScanNode scanNode) throws IOException {
Preconditions.checkNotNull(ctx.getTable(scanNode.getCanonicalName()),
"Error: There is no table matched to %s", scanNode.getCanonicalName() + "(" + scanNode.getTableName() + ")");
@@ -716,8 +775,8 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
return new HashAggregateExec(ctx, groupbyNode, subOp);
}
- private PhysicalExec createSortAggregation(TaskAttemptContext ctx, EnforceProperty property, GroupbyNode groupbyNode, PhysicalExec subOp)
- throws IOException {
+ private PhysicalExec createSortAggregation(TaskAttemptContext ctx, EnforceProperty property, GroupbyNode groupbyNode,
+ PhysicalExec subOp) throws IOException {
Column[] grpColumns = groupbyNode.getGroupingColumns();
SortSpec[] sortSpecs = new SortSpec[grpColumns.length];
@@ -832,6 +891,10 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
type = EnforceType.GROUP_BY;
} else if (node.getType() == NodeType.SORT) {
type = EnforceType.SORT;
+ } else if (node instanceof StoreTableNode
+ && ((StoreTableNode)node).hasPartition()
+ && ((StoreTableNode)node).getPartitionMethod().getPartitionType() == PartitionType.COLUMN) {
+ type = EnforceType.COLUMN_PARTITION;
} else {
return null;
}
@@ -846,6 +909,8 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
found = property;
} else if (type == EnforceType.SORT && property.getSort().getPid() == node.getPID()) {
found = property;
+ } else if (type == EnforceType.COLUMN_PARTITION && property.getColumnPartition().getPid() == node.getPID()) {
+ found = property;
}
}
return found;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/enforce/Enforcer.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/enforce/Enforcer.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/enforce/Enforcer.java
index d7c3ba4..dd46996 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/enforce/Enforcer.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/enforce/Enforcer.java
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import static org.apache.tajo.ipc.TajoWorkerProtocol.*;
+import static org.apache.tajo.ipc.TajoWorkerProtocol.ColumnPartitionEnforcer.ColumnPartitionAlgorithm;
import static org.apache.tajo.ipc.TajoWorkerProtocol.EnforceProperty.EnforceType;
import static org.apache.tajo.ipc.TajoWorkerProtocol.GroupbyEnforce.GroupbyAlgorithm;
@@ -149,6 +150,17 @@ public class Enforcer implements ProtoObject<EnforcerProto> {
TUtil.putToNestedList(properties, builder.getType(), builder.build());
}
+ public void enforceColumnPartitionAlgorithm(int pid, ColumnPartitionAlgorithm algorithm) {
+ EnforceProperty.Builder builder = newProperty();
+ ColumnPartitionEnforcer.Builder enforce = ColumnPartitionEnforcer.newBuilder();
+ enforce.setPid(pid);
+ enforce.setAlgorithm(algorithm);
+
+ builder.setType(EnforceType.COLUMN_PARTITION);
+ builder.setColumnPartition(enforce);
+ TUtil.putToNestedList(properties, builder.getType(), builder.build());
+ }
+
public Collection<EnforceProperty> getProperties() {
if (proto != null) {
return proto.getPropertiesList();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BasicPhysicalExecutorVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BasicPhysicalExecutorVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BasicPhysicalExecutorVisitor.java
index 67d6baa..f6f3e52 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BasicPhysicalExecutorVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/BasicPhysicalExecutorVisitor.java
@@ -31,14 +31,14 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
return visitBNLJoin(context, (BNLJoinExec) exec, stack);
} else if (exec instanceof BSTIndexScanExec) {
return visitBSTIndexScan(context, (BSTIndexScanExec) exec, stack);
- } else if (exec instanceof ColumnPartitionedTableStoreExec) {
- return visitColumnPartitionedTableStore(context, (ColumnPartitionedTableStoreExec) exec, stack);
} else if (exec instanceof EvalExprExec) {
return visitEvalExpr(context, (EvalExprExec) exec, stack);
} else if (exec instanceof ExternalSortExec) {
return visitExternalSort(context, (ExternalSortExec) exec, stack);
} else if (exec instanceof HashAggregateExec) {
return visitHashAggregate(context, (HashAggregateExec) exec, stack);
+ } else if (exec instanceof HashBasedColPartitionStoreExec) {
+ return visitHashBasedColPartitionStore(context, (HashBasedColPartitionStoreExec) exec, stack);
} else if (exec instanceof HashFullOuterJoinExec) {
return visitHashFullOuterJoin(context, (HashFullOuterJoinExec) exec, stack);
} else if (exec instanceof HashJoinExec) {
@@ -77,6 +77,8 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
return visitSeqScan(context, (SeqScanExec) exec, stack);
} else if (exec instanceof SortAggregateExec) {
return visitSortAggregate(context, (SortAggregateExec) exec, stack);
+ } else if (exec instanceof SortBasedColPartitionStoreExec) {
+ return visitSortBasedColPartitionStore(context, (SortBasedColPartitionStoreExec) exec, stack);
} else if (exec instanceof StoreTableExec) {
return visitStoreTable(context, (StoreTableExec) exec, stack);
}
@@ -84,7 +86,7 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
throw new PhysicalPlanningException("Unsupported Type: " + exec.getClass().getSimpleName());
}
- private RESULT visitUnaryExecutor(UnaryPhysicalExec exec, Stack<PhysicalExec> stack, CONTEXT context)
+ private RESULT visitUnaryExecutor(CONTEXT context, UnaryPhysicalExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
stack.push(exec);
RESULT r = visit(exec.getChild(), stack, context);
@@ -92,7 +94,7 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
return r;
}
- private RESULT visitBinaryExecutor(BinaryPhysicalExec exec, Stack<PhysicalExec> stack, CONTEXT context)
+ private RESULT visitBinaryExecutor(CONTEXT context, BinaryPhysicalExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
stack.push(exec);
RESULT r = visit(exec.getLeftChild(), stack, context);
@@ -104,7 +106,7 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
@Override
public RESULT visitBNLJoin(CONTEXT context, BNLJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
@@ -114,11 +116,6 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
}
@Override
- public RESULT visitColumnPartitionedTableStore(CONTEXT context, ColumnPartitionedTableStoreExec exec, Stack<PhysicalExec> stack) throws PhysicalPlanningException {
- return null;
- }
-
- @Override
public RESULT visitEvalExpr(CONTEXT context, EvalExprExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
return null;
@@ -127,115 +124,121 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
@Override
public RESULT visitExternalSort(CONTEXT context, ExternalSortExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitHashAggregate(CONTEXT context, HashAggregateExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitUnaryExecutor(context, exec, stack);
+ }
+
+ @Override
+ public RESULT visitHashBasedColPartitionStore(CONTEXT context, HashBasedColPartitionStoreExec exec,
+ Stack<PhysicalExec> stack) throws PhysicalPlanningException {
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitHashFullOuterJoin(CONTEXT context, HashFullOuterJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitHashJoin(CONTEXT context, HashJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitBinaryExecutor(exec, stack, context);
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitHashLeftAntiJoin(CONTEXT context, HashLeftAntiJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitBinaryExecutor(exec, stack, context);
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitHashLeftOuterJoin(CONTEXT context, HashLeftOuterJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitLeftHashSemiJoin(CONTEXT context, HashLeftSemiJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitBinaryExecutor(exec, stack, context);
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitHashShuffleFileWrite(CONTEXT context, HashShuffleFileWriteExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitHaving(CONTEXT context, HavingExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitLimit(CONTEXT context, LimitExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitUnaryExecutor(exec, stack, context);
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitMemSort(CONTEXT context, MemSortExec exec, Stack<PhysicalExec> stack) throws
PhysicalPlanningException {
- return null;
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitMergeFullOuterJoin(CONTEXT context, MergeFullOuterJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitMergeJoin(CONTEXT context, MergeJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitBinaryExecutor(exec, stack, context);
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitNLJoin(CONTEXT context, NLJoinExec exec, Stack<PhysicalExec> stack) throws
PhysicalPlanningException {
- return null;
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitNLLeftOuterJoin(CONTEXT context, NLLeftOuterJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitProjection(CONTEXT context, ProjectionExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitUnaryExecutor(exec, stack, context);
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitRangeShuffleFileWrite(CONTEXT context, RangeShuffleFileWriteExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitRightOuterMergeJoin(CONTEXT context, RightOuterMergeJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return null;
+ return visitBinaryExecutor(context, exec, stack);
}
@Override
public RESULT visitSelection(CONTEXT context, SelectionExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitUnaryExecutor(exec, stack, context);
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
@@ -246,11 +249,17 @@ public class BasicPhysicalExecutorVisitor<CONTEXT, RESULT> implements PhysicalEx
@Override
public RESULT visitSortAggregate(CONTEXT context, SortAggregateExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException {
- return visitUnaryExecutor(exec, stack, context);
+ return visitUnaryExecutor(context, exec, stack);
+ }
+
+ @Override
+ public RESULT visitSortBasedColPartitionStore(CONTEXT context, SortBasedColPartitionStoreExec exec,
+ Stack<PhysicalExec> stack) throws PhysicalPlanningException {
+ return visitUnaryExecutor(context, exec, stack);
}
@Override
public RESULT visitStoreTable(CONTEXT context, StoreTableExec exec, Stack<PhysicalExec> stack) throws PhysicalPlanningException {
- return null;
+ return visitUnaryExecutor(context, exec, stack);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
new file mode 100644
index 0000000..c81f3bd
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
@@ -0,0 +1,100 @@
+/**
+ * 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.tajo.engine.planner.physical;
+
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.engine.planner.InsertNode;
+import org.apache.tajo.engine.planner.logical.CreateTableNode;
+import org.apache.tajo.engine.planner.logical.NodeType;
+import org.apache.tajo.engine.planner.logical.StoreTableNode;
+import org.apache.tajo.storage.StorageUtil;
+import org.apache.tajo.worker.TaskAttemptContext;
+
+import java.io.IOException;
+
+public abstract class ColPartitionStoreExec extends UnaryPhysicalExec {
+ protected final TableMeta meta;
+ protected final StoreTableNode plan;
+ protected Path storeTablePath;
+
+ protected final int keyNum;
+ protected final int [] keyIds;
+ protected final String [] keyNames;
+
+ public ColPartitionStoreExec(TaskAttemptContext context, StoreTableNode plan, PhysicalExec child) {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child);
+ this.plan = plan;
+
+ if (plan.getType() == NodeType.CREATE_TABLE) {
+ this.outSchema = ((CreateTableNode)plan).getTableSchema();
+ }
+
+ // set table meta
+ if (this.plan.hasOptions()) {
+ meta = CatalogUtil.newTableMeta(plan.getStorageType(), plan.getOptions());
+ } else {
+ meta = CatalogUtil.newTableMeta(plan.getStorageType());
+ }
+
+ // Find column index to name subpartition directory path
+ keyNum = this.plan.getPartitionMethod().getExpressionSchema().getColumnNum();
+
+ keyIds = new int[keyNum];
+ keyNames = new String[keyNum];
+ for (int i = 0; i < keyNum; i++) {
+ Column column = this.plan.getPartitionMethod().getExpressionSchema().getColumn(i);
+ keyNames[i] = column.getColumnName();
+
+ if (this.plan.getType() == NodeType.INSERT) {
+ InsertNode insertNode = ((InsertNode)plan);
+ int idx = insertNode.getTableSchema().getColumnId(column.getQualifiedName());
+ keyIds[i] = idx;
+ } else if (this.plan.getType() == NodeType.CREATE_TABLE) {
+ CreateTableNode createTable = (CreateTableNode) plan;
+ int idx = createTable.getLogicalSchema().getColumnId(column.getQualifiedName());
+ keyIds[i] = idx;
+ } else {
+ // We can get partition column from a logical schema.
+ // Don't use output schema because it is rewritten.
+ keyIds[i] = plan.getOutSchema().getColumnId(column.getQualifiedName());
+ }
+ }
+ }
+
+ @Override
+ public void init() throws IOException {
+ super.init();
+
+ storeTablePath = context.getOutputPath();
+ FileSystem fs = storeTablePath.getFileSystem(context.getConf());
+ if (!fs.exists(storeTablePath.getParent())) {
+ fs.mkdirs(storeTablePath.getParent());
+ }
+ }
+
+
+ protected Path getDataFile(String partition) {
+ return StorageUtil.concatPath(storeTablePath.getParent(), partition, storeTablePath.getName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java
deleted file mode 100644
index cee9bba..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java
+++ /dev/null
@@ -1,192 +0,0 @@
-/**
- * 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.tajo.engine.planner.physical;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.tajo.catalog.CatalogUtil;
-import org.apache.tajo.catalog.Column;
-import org.apache.tajo.catalog.TableMeta;
-import org.apache.tajo.catalog.statistics.StatisticsUtil;
-import org.apache.tajo.catalog.statistics.TableStats;
-import org.apache.tajo.datum.Datum;
-import org.apache.tajo.engine.planner.InsertNode;
-import org.apache.tajo.engine.planner.logical.CreateTableNode;
-import org.apache.tajo.engine.planner.logical.NodeType;
-import org.apache.tajo.engine.planner.logical.StoreTableNode;
-import org.apache.tajo.storage.Appender;
-import org.apache.tajo.storage.StorageManagerFactory;
-import org.apache.tajo.storage.StorageUtil;
-import org.apache.tajo.storage.Tuple;
-import org.apache.tajo.worker.TaskAttemptContext;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * This class is a physical operator to store at column partitioned table.
- */
-public class ColumnPartitionedTableStoreExec extends UnaryPhysicalExec {
- private static Log LOG = LogFactory.getLog(ColumnPartitionedTableStoreExec.class);
-
- private final TableMeta meta;
- private final StoreTableNode plan;
- private Tuple tuple;
- private Path storeTablePath;
- private final Map<String, Appender> appenderMap = new HashMap<String, Appender>();
- private int[] partitionColumnIndices;
- private String[] partitionColumnNames;
-
- public ColumnPartitionedTableStoreExec(TaskAttemptContext context, StoreTableNode plan, PhysicalExec child)
- throws IOException {
- super(context, plan.getInSchema(), plan.getOutSchema(), child);
- this.plan = plan;
-
- if (plan.getType() == NodeType.CREATE_TABLE) {
- this.outSchema = ((CreateTableNode)plan).getTableSchema();
- }
-
- // set table meta
- if (this.plan.hasOptions()) {
- meta = CatalogUtil.newTableMeta(plan.getStorageType(), plan.getOptions());
- } else {
- meta = CatalogUtil.newTableMeta(plan.getStorageType());
- }
-
- // Find column index to name subpartition directory path
- int partitionKeyNum = this.plan.getPartitionMethod().getExpressionSchema().getColumnNum();
- partitionColumnIndices = new int[partitionKeyNum];
- partitionColumnNames = new String[partitionKeyNum];
- for (int i = 0; i < partitionKeyNum; i++) {
- Column column = this.plan.getPartitionMethod().getExpressionSchema().getColumn(i);
- partitionColumnNames[i] = column.getColumnName();
-
- if (this.plan.getType() == NodeType.INSERT) {
- InsertNode insertNode = ((InsertNode)plan);
- int idx = insertNode.getTableSchema().getColumnId(column.getQualifiedName());
- partitionColumnIndices[i] = idx;
- } else if (this.plan.getType() == NodeType.CREATE_TABLE) {
- CreateTableNode createTable = (CreateTableNode) plan;
- int idx = createTable.getLogicalSchema().getColumnId(column.getQualifiedName());
- partitionColumnIndices[i] = idx;
- } else {
- // We can get partition column from a logical schema.
- // Don't use output schema because it is rewritten.
- partitionColumnIndices[i] = plan.getOutSchema().getColumnId(column.getQualifiedName());
- }
- }
- }
-
- public void init() throws IOException {
- super.init();
-
- storeTablePath = context.getOutputPath();
- FileSystem fs = storeTablePath.getFileSystem(context.getConf());
- if (!fs.exists(storeTablePath.getParent())) {
- fs.mkdirs(storeTablePath.getParent());
- }
- }
-
- private Appender getAppender(String partition) throws IOException {
- Appender appender = appenderMap.get(partition);
-
- if (appender == null) {
- Path dataFile = getDataFile(partition);
- FileSystem fs = dataFile.getFileSystem(context.getConf());
-
- if (fs.exists(dataFile.getParent())) {
- LOG.info("Path " + dataFile.getParent() + " already exists!");
- } else {
- fs.mkdirs(dataFile.getParent());
- LOG.info("Add subpartition path directory :" + dataFile.getParent());
- }
-
- if (fs.exists(dataFile)) {
- LOG.info("File " + dataFile + " already exists!");
- FileStatus status = fs.getFileStatus(dataFile);
- LOG.info("File size: " + status.getLen());
- }
-
- appender = StorageManagerFactory.getStorageManager(context.getConf()).getAppender(meta, outSchema, dataFile);
- appender.enableStats();
- appender.init();
- appenderMap.put(partition, appender);
- } else {
- appender = appenderMap.get(partition);
- }
- return appender;
- }
-
- private Path getDataFile(String partition) {
- return StorageUtil.concatPath(storeTablePath.getParent(), partition, storeTablePath.getName());
- }
-
- /* (non-Javadoc)
- * @see PhysicalExec#next()
- */
- @Override
- public Tuple next() throws IOException {
- StringBuilder sb = new StringBuilder();
- while((tuple = child.next()) != null) {
- // set subpartition directory name
- sb.delete(0, sb.length());
- if (partitionColumnIndices != null) {
- for(int i = 0; i < partitionColumnIndices.length; i++) {
- Datum datum = tuple.get(partitionColumnIndices[i]);
- if(i > 0)
- sb.append("/");
- sb.append(partitionColumnNames[i]).append("=");
- sb.append(datum.asChars());
- }
- }
-
- // add tuple
- Appender appender = getAppender(sb.toString());
- appender.addTuple(tuple);
- }
-
- List<TableStats> statSet = new ArrayList<TableStats>();
- for (Map.Entry<String, Appender> entry : appenderMap.entrySet()) {
- Appender app = entry.getValue();
- app.flush();
- app.close();
- statSet.add(app.getStats());
- }
-
- // Collect and aggregated statistics data
- TableStats aggregated = StatisticsUtil.aggregateTableStat(statSet);
- context.setResultStats(aggregated);
-
- return null;
- }
-
- @Override
- public void rescan() throws IOException {
- // nothing to do
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java
new file mode 100644
index 0000000..df32d0b
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/HashBasedColPartitionStoreExec.java
@@ -0,0 +1,132 @@
+/**
+ * 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.tajo.engine.planner.physical;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.statistics.StatisticsUtil;
+import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.engine.planner.logical.StoreTableNode;
+import org.apache.tajo.storage.Appender;
+import org.apache.tajo.storage.StorageManagerFactory;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.worker.TaskAttemptContext;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This class is a physical operator to store at column partitioned table.
+ */
+public class HashBasedColPartitionStoreExec extends ColPartitionStoreExec {
+ private static Log LOG = LogFactory.getLog(HashBasedColPartitionStoreExec.class);
+
+ private final Map<String, Appender> appenderMap = new HashMap<String, Appender>();
+
+ public HashBasedColPartitionStoreExec(TaskAttemptContext context, StoreTableNode plan, PhysicalExec child)
+ throws IOException {
+ super(context, plan, child);
+ }
+
+ public void init() throws IOException {
+ super.init();
+ }
+
+ private Appender getAppender(String partition) throws IOException {
+ Appender appender = appenderMap.get(partition);
+
+ if (appender == null) {
+ Path dataFile = getDataFile(partition);
+ FileSystem fs = dataFile.getFileSystem(context.getConf());
+
+ if (fs.exists(dataFile.getParent())) {
+ LOG.info("Path " + dataFile.getParent() + " already exists!");
+ } else {
+ fs.mkdirs(dataFile.getParent());
+ LOG.info("Add subpartition path directory :" + dataFile.getParent());
+ }
+
+ if (fs.exists(dataFile)) {
+ LOG.info("File " + dataFile + " already exists!");
+ FileStatus status = fs.getFileStatus(dataFile);
+ LOG.info("File size: " + status.getLen());
+ }
+
+ appender = StorageManagerFactory.getStorageManager(context.getConf()).getAppender(meta, outSchema, dataFile);
+ appender.enableStats();
+ appender.init();
+ appenderMap.put(partition, appender);
+ } else {
+ appender = appenderMap.get(partition);
+ }
+ return appender;
+ }
+
+ /* (non-Javadoc)
+ * @see PhysicalExec#next()
+ */
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple;
+ StringBuilder sb = new StringBuilder();
+ while((tuple = child.next()) != null) {
+ // set subpartition directory name
+ sb.delete(0, sb.length());
+ if (keyIds != null) {
+ for(int i = 0; i < keyIds.length; i++) {
+ Datum datum = tuple.get(keyIds[i]);
+ if(i > 0)
+ sb.append("/");
+ sb.append(keyNames[i]).append("=");
+ sb.append(datum.asChars());
+ }
+ }
+
+ // add tuple
+ Appender appender = getAppender(sb.toString());
+ appender.addTuple(tuple);
+ }
+
+ List<TableStats> statSet = new ArrayList<TableStats>();
+ for (Map.Entry<String, Appender> entry : appenderMap.entrySet()) {
+ Appender app = entry.getValue();
+ app.flush();
+ app.close();
+ statSet.add(app.getStats());
+ }
+
+ // Collect and aggregated statistics data
+ TableStats aggregated = StatisticsUtil.aggregateTableStat(statSet);
+ context.setResultStats(aggregated);
+
+ return null;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ // nothing to do
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExecutorVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExecutorVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExecutorVisitor.java
index 9ede15d..738db62 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExecutorVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExecutorVisitor.java
@@ -28,9 +28,6 @@ public interface PhysicalExecutorVisitor<CONTEXT, RESULT> {
RESULT visitBSTIndexScan(CONTEXT context, BSTIndexScanExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException;
- RESULT visitColumnPartitionedTableStore(CONTEXT context, ColumnPartitionedTableStoreExec exec,
- Stack<PhysicalExec> stack) throws PhysicalPlanningException;
-
RESULT visitEvalExpr(CONTEXT context, EvalExprExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException;
@@ -40,6 +37,10 @@ public interface PhysicalExecutorVisitor<CONTEXT, RESULT> {
RESULT visitHashAggregate(CONTEXT context, HashAggregateExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException;
+ RESULT visitHashBasedColPartitionStore(CONTEXT context, HashBasedColPartitionStoreExec exec,
+ Stack<PhysicalExec> stack)
+ throws PhysicalPlanningException;
+
RESULT visitHashFullOuterJoin(CONTEXT context, HashFullOuterJoinExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException;
@@ -97,6 +98,10 @@ public interface PhysicalExecutorVisitor<CONTEXT, RESULT> {
RESULT visitSortAggregate(CONTEXT context, SortAggregateExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException;
+ RESULT visitSortBasedColPartitionStore(CONTEXT context, SortBasedColPartitionStoreExec exec,
+ Stack<PhysicalExec> stack)
+ throws PhysicalPlanningException;
+
RESULT visitStoreTable(CONTEXT context, StoreTableExec exec, Stack<PhysicalExec> stack)
throws PhysicalPlanningException;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java
new file mode 100644
index 0000000..99d673b
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/SortBasedColPartitionStoreExec.java
@@ -0,0 +1,147 @@
+/**
+ * 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.tajo.engine.planner.physical;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.statistics.StatisticsUtil;
+import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.engine.planner.logical.StoreTableNode;
+import org.apache.tajo.storage.Appender;
+import org.apache.tajo.storage.StorageManagerFactory;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.worker.TaskAttemptContext;
+
+import java.io.IOException;
+
+/**
+ * It stores a sorted data set into a number of partition files. It assumes that input tuples are sorted in an
+ * ascending or descending order of partition columns.
+ */
+public class SortBasedColPartitionStoreExec extends ColPartitionStoreExec {
+ private static Log LOG = LogFactory.getLog(SortBasedColPartitionStoreExec.class);
+
+ private Tuple currentKey;
+ private Tuple prevKey;
+
+ private Appender appender;
+ private TableStats aggregated;
+
+ public SortBasedColPartitionStoreExec(TaskAttemptContext context, StoreTableNode plan, PhysicalExec child)
+ throws IOException {
+ super(context, plan, child);
+ }
+
+ public void init() throws IOException {
+ super.init();
+
+ currentKey = new VTuple(keyNum);
+ aggregated = new TableStats();
+ }
+
+ private Appender getAppender(String partition) throws IOException {
+ Path dataFile = getDataFile(partition);
+ FileSystem fs = dataFile.getFileSystem(context.getConf());
+
+ if (fs.exists(dataFile.getParent())) {
+ LOG.info("Path " + dataFile.getParent() + " already exists!");
+ } else {
+ fs.mkdirs(dataFile.getParent());
+ LOG.info("Add subpartition path directory :" + dataFile.getParent());
+ }
+
+ if (fs.exists(dataFile)) {
+ LOG.info("File " + dataFile + " already exists!");
+ FileStatus status = fs.getFileStatus(dataFile);
+ LOG.info("File size: " + status.getLen());
+ }
+
+ appender = StorageManagerFactory.getStorageManager(context.getConf()).getAppender(meta, outSchema, dataFile);
+ appender.enableStats();
+ appender.init();
+
+ return appender;
+ }
+
+ private void fillKeyTuple(Tuple inTuple, Tuple keyTuple) {
+ for (int i = 0; i < keyIds.length; i++) {
+ keyTuple.put(i, inTuple.get(keyIds[i]));
+ }
+ }
+
+ private String getSubdirectory(Tuple keyTuple) {
+ StringBuilder sb = new StringBuilder();
+
+ for(int i = 0; i < keyIds.length; i++) {
+ Datum datum = keyTuple.get(i);
+ if(i > 0) {
+ sb.append("/");
+ }
+ sb.append(keyNames[i]).append("=");
+ sb.append(datum.asChars());
+ }
+ return sb.toString();
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ Tuple tuple;
+ while((tuple = child.next()) != null) {
+
+ fillKeyTuple(tuple, currentKey);
+
+ if (prevKey == null) {
+ appender = getAppender(getSubdirectory(currentKey));
+ prevKey = new VTuple(currentKey);
+ } else {
+ if (!prevKey.equals(currentKey)) {
+ appender.close();
+ StatisticsUtil.aggregateTableStat(aggregated, appender.getStats());
+
+ appender = getAppender(getSubdirectory(currentKey));
+ prevKey = new VTuple(currentKey);
+ }
+ }
+
+ appender.addTuple(tuple);
+ }
+
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ appender.close();
+ StatisticsUtil.aggregateTableStat(aggregated, appender.getStats());
+ context.setResultStats(aggregated);
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ // nothing to do
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
index 9aa6d86..e08da29 100644
--- a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
+++ b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
@@ -170,6 +170,7 @@ message EnforceProperty {
JOIN = 3;
SORT = 4;
BROADCAST = 5;
+ COLUMN_PARTITION = 6;
}
// Identifies which field is filled in.
@@ -182,6 +183,7 @@ message EnforceProperty {
optional JoinEnforce join = 5;
optional SortEnforce sort = 6;
optional BroadcastEnforce broadcast = 7;
+ optional ColumnPartitionEnforcer columnPartition = 8;
}
message SortedInputEnforce {
@@ -230,6 +232,16 @@ message BroadcastEnforce {
required string tableName = 1;
}
+message ColumnPartitionEnforcer {
+ enum ColumnPartitionAlgorithm {
+ HASH_PARTITION = 0;
+ SORT_PARTITION = 1;
+ }
+
+ required int32 pid = 1;
+ required ColumnPartitionAlgorithm algorithm = 2;
+}
+
message EnforcerProto {
repeated EnforceProperty properties = 1;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/10c599f4/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
----------------------------------------------------------------------
diff --git 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
index ab27a45..7cdf6ec 100644
--- 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
@@ -63,6 +63,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static org.apache.tajo.ipc.TajoWorkerProtocol.ColumnPartitionEnforcer.ColumnPartitionAlgorithm;
import static org.apache.tajo.ipc.TajoWorkerProtocol.ShuffleType;
import static org.apache.tajo.ipc.TajoWorkerProtocol.SortEnforce.SortAlgorithm;
import static org.junit.Assert.*;
@@ -350,8 +351,9 @@ public class TestPhysicalPlanner {
}
private String[] CreateTableAsStmts = {
- "create table grouped1 as select deptName, class, sum(score), max(score), min(score) from score group by deptName, class", // 8
- "create table grouped2 using rcfile as select deptName, class, sum(score), max(score), min(score) from score group by deptName, class", // 8
+ "create table grouped1 as select deptName, class, sum(score), max(score), min(score) from score group by deptName, class", // 0
+ "create table grouped2 using rcfile as select deptName, class, sum(score), max(score), min(score) from score group by deptName, class", // 1
+ "create table grouped3 partition by column (dept text, class text) as select sum(score), max(score), min(score), deptName, class from score group by deptName, class", // 2
};
@Test
@@ -436,6 +438,70 @@ public class TestPhysicalPlanner {
}
@Test
+ public final void testEnforceForDefaultColumnPartitionStorePlan() throws IOException, PlanningException {
+ FileFragment[] frags = StorageManager.splitNG(conf, "score", score.getMeta(), score.getPath(),
+ Integer.MAX_VALUE);
+ Path workDir = CommonTestingUtil.getTestDir("target/test-data/testStorePlan");
+ TaskAttemptContext ctx = new TaskAttemptContext(conf, LocalTajoTestingUtility.newQueryUnitAttemptId(masterPlan),
+ new FileFragment[] { frags[0] }, workDir);
+ ctx.setEnforcer(new Enforcer());
+ ctx.setOutputPath(new Path(workDir, "grouped3"));
+
+ Expr context = analyzer.parse(CreateTableAsStmts[2]);
+ LogicalPlan plan = planner.createPlan(context);
+ LogicalNode rootNode = optimizer.optimize(plan);
+ PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf,sm);
+ PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
+ assertTrue(exec instanceof SortBasedColPartitionStoreExec);
+ }
+
+ @Test
+ public final void testEnforceForHashBasedColumnPartitionStorePlan() throws IOException, PlanningException {
+
+ Expr context = analyzer.parse(CreateTableAsStmts[2]);
+ LogicalPlan plan = planner.createPlan(context);
+ LogicalRootNode rootNode = (LogicalRootNode) optimizer.optimize(plan);
+ CreateTableNode createTableNode = rootNode.getChild();
+ Enforcer enforcer = new Enforcer();
+ enforcer.enforceColumnPartitionAlgorithm(createTableNode.getPID(), ColumnPartitionAlgorithm.HASH_PARTITION);
+
+ FileFragment[] frags = StorageManager.splitNG(conf, "score", score.getMeta(), score.getPath(),
+ Integer.MAX_VALUE);
+ Path workDir = CommonTestingUtil.getTestDir("target/test-data/testStorePlan");
+ TaskAttemptContext ctx = new TaskAttemptContext(conf, LocalTajoTestingUtility.newQueryUnitAttemptId(masterPlan),
+ new FileFragment[] { frags[0] }, workDir);
+ ctx.setEnforcer(enforcer);
+ ctx.setOutputPath(new Path(workDir, "grouped4"));
+
+ PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf,sm);
+ PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
+ assertTrue(exec instanceof HashBasedColPartitionStoreExec);
+ }
+
+ @Test
+ public final void testEnforceForSortBasedColumnPartitionStorePlan() throws IOException, PlanningException {
+
+ Expr context = analyzer.parse(CreateTableAsStmts[2]);
+ LogicalPlan plan = planner.createPlan(context);
+ LogicalRootNode rootNode = (LogicalRootNode) optimizer.optimize(plan);
+ CreateTableNode createTableNode = rootNode.getChild();
+ Enforcer enforcer = new Enforcer();
+ enforcer.enforceColumnPartitionAlgorithm(createTableNode.getPID(), ColumnPartitionAlgorithm.SORT_PARTITION);
+
+ FileFragment[] frags = StorageManager.splitNG(conf, "score", score.getMeta(), score.getPath(),
+ Integer.MAX_VALUE);
+ Path workDir = CommonTestingUtil.getTestDir("target/test-data/testStorePlan");
+ TaskAttemptContext ctx = new TaskAttemptContext(conf, LocalTajoTestingUtility.newQueryUnitAttemptId(masterPlan),
+ new FileFragment[] { frags[0] }, workDir);
+ ctx.setEnforcer(enforcer);
+ ctx.setOutputPath(new Path(workDir, "grouped5"));
+
+ PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf,sm);
+ PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
+ assertTrue(exec instanceof SortBasedColPartitionStoreExec);
+ }
+
+ @Test
public final void testPartitionedStorePlan() throws IOException, PlanningException {
FileFragment[] frags = StorageManager.splitNG(conf, "score", score.getMeta(), score.getPath(),
Integer.MAX_VALUE);