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 2013/09/16 13:33:34 UTC
[7/7] git commit: TAJO-184: Refactor GlobalPlanner and global plan
data structure. (hyunsik)
TAJO-184: Refactor GlobalPlanner and global plan data structure. (hyunsik)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/1b1d1e8c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/1b1d1e8c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/1b1d1e8c
Branch: refs/heads/master
Commit: 1b1d1e8c1a6b82ccc5c3ce4daeb9e3daa309cde4
Parents: 53df7f4
Author: Hyunsik Choi <hy...@apache.org>
Authored: Mon Sep 16 20:32:05 2013 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Mon Sep 16 20:32:05 2013 +0900
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../org/apache/tajo/catalog/CatalogUtil.java | 23 +-
.../java/org/apache/tajo/catalog/Column.java | 72 +-
.../java/org/apache/tajo/catalog/Schema.java | 96 +-
.../tajo/catalog/statistics/StatisticsUtil.java | 29 +
.../src/main/proto/CatalogProtos.proto | 9 +-
.../org/apache/tajo/catalog/TestColumn.java | 14 +-
.../org/apache/tajo/catalog/TestSchema.java | 15 +-
.../org/apache/tajo/catalog/CatalogServer.java | 13 +-
.../java/org/apache/tajo/ExecutionBlockId.java | 10 -
.../java/org/apache/tajo/QueryIdFactory.java | 8 -
.../main/java/org/apache/tajo/util/TUtil.java | 35 +-
.../tajo-core-backend/benchmark/tpch/q12.tql | 10 +-
.../tajo-core-backend/benchmark/tpch/q13.tql | 23 +-
.../tajo-core-backend/benchmark/tpch/q14.tql | 10 +-
.../tajo-core-backend/benchmark/tpch/q17.tql | 2 +-
.../tajo-core-backend/benchmark/tpch/q2.tql | 58 +-
.../tajo-core-backend/benchmark/tpch/q4.tql | 26 +-
.../tajo-core-backend/benchmark/tpch/q7.tql | 57 +-
.../tajo-core-backend/benchmark/tpch/q8.tql | 60 +-
.../tajo-core-backend/benchmark/tpch/q9.tql | 50 +-
.../main/java/org/apache/tajo/DataChannel.java | 198 ++++
.../org/apache/tajo/TaskAttemptContext.java | 9 +
.../apache/tajo/engine/eval/EvalTreeUtil.java | 10 +-
.../org/apache/tajo/engine/eval/FieldEval.java | 4 +-
.../apache/tajo/engine/parser/SQLAnalyzer.java | 13 +-
.../tajo/engine/planner/AlgebraVisitor.java | 1 +
.../tajo/engine/planner/BaseAlgebraVisitor.java | 11 +
.../engine/planner/BasicLogicalPlanVisitor.java | 12 +
.../planner/ExplainLogicalPlanVisitor.java | 202 ++++
.../apache/tajo/engine/planner/InsertNode.java | 12 +
.../apache/tajo/engine/planner/LogicalPlan.java | 139 ++-
.../tajo/engine/planner/LogicalPlanVisitor.java | 2 +
.../tajo/engine/planner/LogicalPlanner.java | 48 +-
.../engine/planner/PhysicalPlannerImpl.java | 77 +-
.../apache/tajo/engine/planner/PlanString.java | 99 ++
.../apache/tajo/engine/planner/PlannerUtil.java | 141 +--
.../apache/tajo/engine/planner/Projector.java | 4 +-
.../tajo/engine/planner/QueryBlockGraph.java | 91 --
.../org/apache/tajo/engine/planner/Target.java | 10 +-
.../engine/planner/global/GlobalOptimizer.java | 80 --
.../tajo/engine/planner/global/MasterPlan.java | 217 ++++-
.../engine/planner/graph/DirectedGraph.java | 65 ++
.../planner/graph/DirectedGraphVisitor.java | 25 +
.../planner/graph/SimpleDirectedGraph.java | 219 +++++
.../tajo/engine/planner/logical/BinaryNode.java | 18 +-
.../engine/planner/logical/CreateTableNode.java | 7 +
.../engine/planner/logical/DropTableNode.java | 7 +
.../engine/planner/logical/EvalExprNode.java | 6 +
.../tajo/engine/planner/logical/ExceptNode.java | 11 +
.../engine/planner/logical/GroupbyNode.java | 34 +
.../engine/planner/logical/IndexScanNode.java | 2 -
.../engine/planner/logical/IntersectNode.java | 11 +
.../tajo/engine/planner/logical/JoinNode.java | 13 +-
.../tajo/engine/planner/logical/LimitNode.java | 12 +-
.../engine/planner/logical/LogicalNode.java | 6 +-
.../engine/planner/logical/LogicalRootNode.java | 7 +
.../tajo/engine/planner/logical/NodeType.java | 4 +-
.../engine/planner/logical/ProjectionNode.java | 28 +
.../engine/planner/logical/ReceiveNode.java | 10 +-
.../engine/planner/logical/RelationNode.java | 35 +
.../tajo/engine/planner/logical/ScanNode.java | 60 +-
.../engine/planner/logical/SelectionNode.java | 28 +-
.../tajo/engine/planner/logical/SendNode.java | 8 +-
.../tajo/engine/planner/logical/SortNode.java | 18 +-
.../engine/planner/logical/StoreTableNode.java | 32 +-
.../planner/logical/TableSubQueryNode.java | 123 +++
.../tajo/engine/planner/logical/UnaryNode.java | 7 +-
.../tajo/engine/planner/logical/UnionNode.java | 12 +
.../engine/planner/logical/join/JoinTree.java | 4 +-
.../physical/BasicPhysicalExecutorVisitor.java | 127 +++
.../physical/PhysicalExecutorVisitor.java | 37 +
.../planner/physical/PhysicalPlanUtil.java | 40 +
.../physical/PhysicalPlanningException.java | 31 +
.../engine/planner/physical/SeqScanExec.java | 2 +-
.../planner/rewrite/FilterPushDownRule.java | 3 +-
.../planner/rewrite/ProjectionPushDownRule.java | 137 ++-
.../tajo/engine/query/QueryUnitRequestImpl.java | 33 +-
.../engine/query/ResultSetMetaDataImpl.java | 2 +-
.../ipc/protocolrecords/QueryUnitRequest.java | 2 +
.../org/apache/tajo/master/ExecutionBlock.java | 53 +-
.../tajo/master/ExecutionBlockCursor.java | 24 +-
.../org/apache/tajo/master/GlobalEngine.java | 27 +-
.../org/apache/tajo/master/GlobalPlanner.java | 951 +++++++------------
.../org/apache/tajo/master/QueryContext.java | 8 +
.../apache/tajo/master/TajoAsyncDispatcher.java | 2 -
.../apache/tajo/master/TaskSchedulerImpl.java | 26 +-
.../tajo/master/event/QueryStartEvent.java | 8 +-
.../apache/tajo/master/querymaster/Query.java | 52 +-
.../master/querymaster/QueryInProgress.java | 1 +
.../tajo/master/querymaster/QueryMaster.java | 42 +-
.../master/querymaster/QueryMasterTask.java | 64 +-
.../tajo/master/querymaster/QueryUnit.java | 5 +-
.../tajo/master/querymaster/Repartitioner.java | 260 +++--
.../tajo/master/querymaster/SubQuery.java | 133 ++-
.../tajo/worker/TajoWorkerManagerService.java | 3 +-
.../main/java/org/apache/tajo/worker/Task.java | 10 +-
.../src/main/proto/TajoWorkerProtocol.proto | 32 +-
.../resources/webapps/worker/querydetail.jsp | 39 +-
.../apache/tajo/LocalTajoTestingUtility.java | 18 +
.../org/apache/tajo/TestQueryIdFactory.java | 17 +-
.../test/java/org/apache/tajo/TestTajoIds.java | 13 +-
.../tajo/engine/parser/TestSQLAnalyzer.java | 12 +
.../plan/global/TestGlobalQueryPlanner.java | 391 --------
.../planner/TestGenericDirectedGraph.java | 73 ++
.../tajo/engine/planner/TestLogicalPlan.java | 116 +++
.../tajo/engine/planner/TestLogicalPlanner.java | 268 +++---
.../tajo/engine/planner/TestPlannerUtil.java | 25 +-
.../global/TestGlobalQueryOptimizer.java | 171 ----
.../engine/planner/global/TestMasterPlan.java | 54 ++
.../planner/physical/TestBNLJoinExec.java | 5 +-
.../planner/physical/TestBSTIndexExec.java | 12 +-
.../planner/physical/TestExternalSortExec.java | 4 +-
.../planner/physical/TestHashAntiJoinExec.java | 3 +-
.../planner/physical/TestHashJoinExec.java | 3 +-
.../planner/physical/TestHashSemiJoinExec.java | 3 +-
.../planner/physical/TestMergeJoinExec.java | 3 +-
.../engine/planner/physical/TestNLJoinExec.java | 10 +-
.../planner/physical/TestPhysicalPlanner.java | 86 +-
.../engine/planner/physical/TestSortExec.java | 4 +-
.../tajo/engine/query/TestGroupByQuery.java | 37 +-
.../tajo/engine/query/TestResultSetImpl.java | 2 +-
.../tajo/engine/query/TestSelectQuery.java | 5 +-
.../tajo/engine/query/TestTableSubQuery.java | 73 ++
.../tajo/master/TestExecutionBlockCursor.java | 14 +-
.../apache/tajo/master/TestGlobalPlanner.java | 185 ++++
.../apache/tajo/master/TestRepartitioner.java | 4 +-
.../tajo/worker/TestRangeRetrieverHandler.java | 6 +-
.../worker/dataserver/TestHttpDataServer.java | 10 +-
.../src/test/queries/complex_union_1.sql | 29 +
.../src/test/queries/complex_union_2.sql | 35 +
.../src/test/queries/select_13.hiveql | 2 +-
.../src/test/queries/set_4.sql | 36 +
.../src/test/queries/table_subquery2.sql | 21 +
.../src/test/resources/catalog-default.xml | 2 +-
.../tajo/storage/TestTupleComparator.java | 4 +-
.../apache/tajo/storage/index/TestBSTIndex.java | 60 +-
.../index/TestSingleCSVFileBSTIndex.java | 8 +-
138 files changed, 4264 insertions(+), 2528 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 73c3b61..58b90a9 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -25,6 +25,8 @@ Release 0.2.0 - unreleased
IMPROVEMENTS
+ TAJO-184: Refactor GlobalPlanner and global plan data structure. (hyunsik)
+
TAJO-119: Refactor and Improve Datum. (jinho)
TAJO-178: Implements StorageManager for scanning asynchronously.
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
index 8f17852..fb55bf7 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
@@ -141,27 +141,14 @@ public class CatalogUtil {
*
* @return
*/
- public static SchemaProto getQualfiedSchema(String tableName,
- SchemaProto schema) {
+ public static SchemaProto getQualfiedSchema(String tableName, SchemaProto schema) {
SchemaProto.Builder revisedSchema = SchemaProto.newBuilder(schema);
revisedSchema.clearFields();
- String[] split;
for (ColumnProto col : schema.getFieldsList()) {
- split = col.getColumnName().split("\\.");
- if (split.length == 1) { // if not qualified name
- // rewrite the column
- ColumnProto.Builder builder = ColumnProto.newBuilder(col);
- builder.setColumnName(tableName + "." + col.getColumnName());
- col = builder.build();
- } else if (split.length == 2) {
- ColumnProto.Builder builder = ColumnProto.newBuilder(col);
- builder.setColumnName(tableName + "." + split[1]);
- col = builder.build();
- } else {
- throw new InternalError("Unaccetable field name "
- + col.getColumnName());
- }
- revisedSchema.addFields(col);
+ ColumnProto.Builder builder = ColumnProto.newBuilder(col);
+ builder.setColumnName(col.getColumnName());
+ builder.setQualifier(tableName);
+ revisedSchema.addFields(builder.build());
}
return revisedSchema.build();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Column.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Column.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Column.java
index 2227999..1b623fa 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Column.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Column.java
@@ -18,17 +18,20 @@
package org.apache.tajo.catalog;
+import com.google.common.base.Objects;
import com.google.gson.annotations.Expose;
-import org.apache.tajo.json.GsonObject;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
import org.apache.tajo.common.ProtoObject;
import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.common.TajoDataTypes.DataType;
+import org.apache.tajo.json.GsonObject;
+import org.apache.tajo.util.TUtil;
public class Column implements ProtoObject<ColumnProto>, Cloneable, GsonObject {
private ColumnProto.Builder builder = null;
-
+
+ @Expose protected String qualifier; // optional
@Expose protected String name; // required
@Expose protected DataType dataType; // required
@@ -38,7 +41,7 @@ public class Column implements ProtoObject<ColumnProto>, Cloneable, GsonObject {
public Column(String columnName, DataType dataType) {
this();
- this.name = columnName.toLowerCase();
+ checkAndSetName(columnName.toLowerCase());
this.dataType = dataType;
}
@@ -51,34 +54,55 @@ public class Column implements ProtoObject<ColumnProto>, Cloneable, GsonObject {
}
public Column(ColumnProto proto) {
- this(proto.getColumnName(), proto.getDataType());
+ this();
+ name = proto.getColumnName();
+ dataType = proto.getDataType();
+ if (proto.hasQualifier()) {
+ qualifier = proto.getQualifier();
+ }
}
+ private void checkAndSetName(String qualifiedOrName) {
+ String [] splits = qualifiedOrName.split("\\.");
+ if (splits.length > 1) {
+ qualifier = qualifiedOrName.substring(0, qualifiedOrName.lastIndexOf("."));
+ name = qualifiedOrName.substring(qualifiedOrName.lastIndexOf(".") + 1, qualifiedOrName.length());
+ } else {
+ qualifier = null;
+ name = qualifiedOrName;
+ }
+ }
+
public String getQualifiedName() {
- return this.name;
+ if (qualifier != null) {
+ return qualifier + "." + name;
+ } else {
+ return name;
+ }
}
- public boolean isQualified() {
- return getQualifiedName().split("\\.").length == 2;
+ public boolean hasQualifier() {
+ return qualifier != null;
}
- public String getTableName() {
- if (isQualified()) {
- return getQualifiedName().split("\\.")[0];
+ public void setQualifier(String qualifier) {
+ this.qualifier = qualifier;
+ }
+
+ public String getQualifier() {
+ if (qualifier != null) {
+ return qualifier;
} else {
return "";
}
}
public String getColumnName() {
- if (isQualified())
- return getQualifiedName().split("\\.")[1];
- else
- return getQualifiedName();
+ return name;
}
public void setName(String name) {
- this.name = name.toLowerCase();
+ checkAndSetName(name.toLowerCase());
}
public DataType getDataType() {
@@ -92,15 +116,17 @@ public class Column implements ProtoObject<ColumnProto>, Cloneable, GsonObject {
@Override
public boolean equals(Object o) {
if (o instanceof Column) {
- Column cd = (Column)o;
- return this.getQualifiedName().equals(cd.getQualifiedName()) &&
- this.getDataType().equals(cd.getDataType());
- }
+ Column another = (Column)o;
+ return name.equals(another.name) &&
+ dataType.equals(another.dataType) &&
+ TUtil.checkEquals(qualifier, another.qualifier);
+ }
return false;
}
public int hashCode() {
- return getQualifiedName().hashCode() ^ (getDataType().hashCode() * 17);
+ return Objects.hashCode(name, dataType, qualifier);
+
}
@Override
@@ -109,6 +135,7 @@ public class Column implements ProtoObject<ColumnProto>, Cloneable, GsonObject {
column.builder = ColumnProto.newBuilder();
column.name = name;
column.dataType = dataType;
+ column.qualifier = qualifier != null ? qualifier : null;
return column;
}
@@ -116,12 +143,15 @@ public class Column implements ProtoObject<ColumnProto>, Cloneable, GsonObject {
public ColumnProto getProto() {
builder.setColumnName(this.name);
builder.setDataType(this.dataType);
+ if (qualifier != null) {
+ builder.setQualifier(qualifier);
+ }
return builder.build();
}
public String toString() {
- return getQualifiedName() +" (" + getDataType().getType() +")";
+ return getQualifiedName() +" (" + getDataType().getType() +"(" + getDataType().getLength() + "))";
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
index 65f6176..4659a02 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java
@@ -21,7 +21,6 @@ package org.apache.tajo.catalog;
import com.google.gson.annotations.Expose;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.tajo.json.GsonObject;
import org.apache.tajo.catalog.exception.AlreadyExistsFieldException;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
@@ -29,6 +28,8 @@ import org.apache.tajo.catalog.proto.CatalogProtos.SchemaProto;
import org.apache.tajo.common.ProtoObject;
import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.json.GsonObject;
+import org.apache.tajo.util.TUtil;
import java.util.*;
@@ -37,26 +38,39 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
private SchemaProto.Builder builder = SchemaProto.newBuilder();
@Expose protected List<Column> fields = null;
- @Expose protected Map<String,Integer> fieldsByName = null;
+ @Expose protected Map<String, Integer> fieldsByQialifiedName = null;
+ @Expose protected Map<String, List<Integer>> fieldsByName = null;
public Schema() {
this.fields = new ArrayList<Column>();
- this.fieldsByName = new TreeMap<String, Integer>();
+ this.fieldsByQialifiedName = new TreeMap<String, Integer>();
+ this.fieldsByName = new HashMap<String, List<Integer>>();
}
public Schema(SchemaProto proto) {
this.fields = new ArrayList<Column>();
- this.fieldsByName = new HashMap<String, Integer>();
+ this.fieldsByQialifiedName = new HashMap<String, Integer>();
+ this.fieldsByName = new HashMap<String, List<Integer>>();
for(ColumnProto colProto : proto.getFieldsList()) {
fields.add(new Column(colProto));
- fieldsByName.put(colProto.getColumnName(), fields.size() - 1);
+ if (colProto.hasQualifier()) {
+ fieldsByQialifiedName.put(colProto.getQualifier() + "." + colProto.getColumnName(), fields.size() - 1);
+ } else {
+ fieldsByQialifiedName.put(colProto.getColumnName(), fields.size() - 1);
+ }
+ if (fieldsByName.containsKey(colProto.getColumnName())) {
+ fieldsByName.get(colProto.getColumnName()).add(fields.size() - 1);
+ } else {
+ fieldsByName.put(colProto.getColumnName(), TUtil.newList(fields.size() - 1));
+ }
}
}
public Schema(Schema schema) {
this();
this.fields.addAll(schema.fields);
- this.fieldsByName.putAll(schema.fieldsByName);
+ this.fieldsByQialifiedName.putAll(schema.fieldsByQialifiedName);
+ this.fieldsByName.putAll(schema.fieldsByName);
}
public Schema(Column [] columns) {
@@ -65,23 +79,48 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
addColumn(c);
}
}
+
+ public void setQualifier(String qualifier) {
+ fieldsByQialifiedName.clear();
+
+ for (int i = 0; i < getColumnNum(); i++) {
+ fields.get(i).setQualifier(qualifier);
+ fieldsByQialifiedName.put(fields.get(i).getQualifiedName(), i);
+ }
+ }
public int getColumnNum() {
return this.fields.size();
}
- public Column getColumn(String colName) {
- Integer cid = fieldsByName.get(colName.toLowerCase());
+ public Column getColumnByFQN(String colName) {
+ Integer cid = fieldsByQialifiedName.get(colName.toLowerCase());
return cid != null ? fields.get(cid) : null;
}
public Column getColumnByName(String colName) {
- for (Column col : fields) {
- if (col.getColumnName().equals(colName.toLowerCase())) {
- return col;
- }
- }
- return null;
+ String normalized = colName.toLowerCase();
+ List<Integer> list = fieldsByName.get(normalized);
+
+ if (list == null || list.size() == 0) {
+ return null;
+ }
+
+ if (list.size() == 1) {
+ return fields.get(list.get(0));
+ } else {
+ StringBuilder sb = new StringBuilder();
+ boolean first = true;
+ for (Integer id : list) {
+ if (first) {
+ first = false;
+ } else {
+ sb.append(", ");
+ }
+ sb.append(fields.get(id));
+ }
+ throw new RuntimeException("Ambiguous Column Name: " + sb.toString());
+ }
}
public Column getColumn(int id) {
@@ -89,13 +128,13 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
}
public int getColumnId(String colName) {
- return fieldsByName.get(colName.toLowerCase());
+ return fieldsByQialifiedName.get(colName.toLowerCase());
}
public int getColumnIdByName(String colName) {
for (Column col : fields) {
if (col.getColumnName().equals(colName.toLowerCase())) {
- return fieldsByName.get(col.getQualifiedName());
+ return fieldsByQialifiedName.get(col.getQualifiedName());
}
}
return -1;
@@ -110,7 +149,7 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
}
public boolean contains(String colName) {
- return fieldsByName.containsKey(colName.toLowerCase());
+ return fieldsByQialifiedName.containsKey(colName.toLowerCase());
}
public synchronized Schema addColumn(String name, Type type) {
@@ -125,21 +164,22 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
}
public synchronized Schema addColumn(String name, DataType dataType) {
- String lowcased = name.toLowerCase();
- if(fieldsByName.containsKey(lowcased)) {
- LOG.error("Already exists column " + lowcased);
- throw new AlreadyExistsFieldException(lowcased);
+ String nomalized = name.toLowerCase();
+ if(fieldsByQialifiedName.containsKey(nomalized)) {
+ LOG.error("Already exists column " + nomalized);
+ throw new AlreadyExistsFieldException(nomalized);
}
- Column newCol = new Column(lowcased, dataType);
+ Column newCol = new Column(nomalized, dataType);
fields.add(newCol);
- fieldsByName.put(lowcased, fields.size() - 1);
+ fieldsByQialifiedName.put(newCol.getQualifiedName(), fields.size() - 1);
+ fieldsByName.put(newCol.getColumnName(), TUtil.newList(fields.size() - 1));
return this;
}
public synchronized void addColumn(Column column) {
- addColumn(column.getQualifiedName(), column.getDataType());
+ addColumn(column.getQualifiedName(), column.getDataType());
}
public synchronized void addColumns(Schema schema) {
@@ -158,12 +198,8 @@ public class Schema implements ProtoObject<SchemaProto>, Cloneable, GsonObject {
}
@Override
- public Object clone() throws CloneNotSupportedException {
- Schema schema = (Schema) super.clone();
- schema.builder = SchemaProto.newBuilder();
- schema.fields = fields != null ? new ArrayList<Column>(fields) : null;
- schema.fieldsByName = fieldsByName != null ? new TreeMap<String, Integer>(fieldsByName) : null;
-
+ public Object clone() {
+ Schema schema = new Schema(toArray());
return schema;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/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 188fdcc..470e1a5 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
@@ -22,6 +22,8 @@ import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
public class StatisticsUtil {
@@ -101,4 +103,31 @@ public class StatisticsUtil {
return aggregated;
}
+
+ public static TableStat computeStatFromUnionBlock(Collection<TableStat> stats) {
+ TableStat stat = new TableStat();
+ TableStat childStat;
+ long avgRows = 0, numBytes = 0, numRows = 0;
+ int numBlocks = 0, numPartitions = 0;
+ List<ColumnStat> columnStats = Lists.newArrayList();
+
+ Iterator<TableStat> it = stats.iterator();
+ while (it.hasNext()) {
+ childStat = it.next();
+ avgRows += childStat.getAvgRows();
+ columnStats.addAll(childStat.getColumnStats());
+ numBlocks += childStat.getNumBlocks();
+ numBytes += childStat.getNumBytes();
+ numPartitions += childStat.getNumPartitions();
+ numRows += childStat.getNumRows();
+ }
+
+ stat.setColumnStats(columnStats);
+ stat.setNumBlocks(numBlocks);
+ stat.setNumBytes(numBytes);
+ stat.setNumPartitions(numPartitions);
+ stat.setNumRows(numRows);
+ stat.setAvgRows(avgRows);
+ return stat;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
index 3d61ebb..477362c 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -64,7 +64,8 @@ message ColumnMetaProto {
message ColumnProto {
required string columnName = 1;
- required DataType dataType = 2;
+ optional string qualifier = 2;
+ required DataType dataType = 3;
}
message SchemaProto {
@@ -130,9 +131,9 @@ message IndexDescProto {
enum IndexMethod {
TWO_LEVEL_BIN_TREE = 0;
- BTREE = 1;
- HASH = 2;
- BITMAP = 3;
+ BTREE_IDX = 1;
+ HASH_IDX = 2;
+ BITMAP_IDX = 3;
}
message GetAllTableNamesResponse {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestColumn.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestColumn.java b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestColumn.java
index dfea828..406149e 100644
--- a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestColumn.java
+++ b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestColumn.java
@@ -77,12 +77,22 @@ public class TestColumn {
public final void testQualifiedName() {
Column col = new Column("table_1.id", Type.INT4);
- assertTrue(col.isQualified());
+ assertTrue(col.hasQualifier());
assertEquals("id", col.getColumnName());
assertEquals("table_1.id", col.getQualifiedName());
- assertEquals("table_1", col.getTableName());
+ assertEquals("table_1", col.getQualifier());
}
+ @Test
+ public final void testMultiLevelQualifiedName() {
+ Column col = new Column("database1.table_1.id", Type.INT4);
+
+ assertTrue(col.hasQualifier());
+ assertEquals("id", col.getColumnName());
+ assertEquals("database1.table_1.id", col.getQualifiedName());
+ assertEquals("database1.table_1", col.getQualifier());
+ }
+
@Test
public final void testToJson() {
Column col = new Column(field1.getProto());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestSchema.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestSchema.java b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestSchema.java
index 689f004..e1be420 100644
--- a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestSchema.java
+++ b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestSchema.java
@@ -61,9 +61,9 @@ public class TestSchema {
@Test
public final void testGetColumnString() {
- assertEquals(col1, schema.getColumn("name"));
- assertEquals(col2, schema.getColumn("age"));
- assertEquals(col3, schema.getColumn("addr"));
+ assertEquals(col1, schema.getColumnByFQN("name"));
+ assertEquals(col2, schema.getColumnByFQN("age"));
+ assertEquals(col3, schema.getColumnByFQN("addr"));
}
@Test
@@ -134,4 +134,13 @@ public class TestSchema {
Schema fromJson = new Schema(proto);
assertEquals(schema2, fromJson);
}
+
+ @Test
+ public final void testSetQualifier() {
+ Schema schema2 = new Schema(schema.getProto());
+ schema2.setQualifier("test1");
+ Column column = schema2.getColumn(1);
+ assertEquals(column, schema2.getColumnByName("age"));
+ assertEquals(column, schema2.getColumnByFQN("test1.age"));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
index 3ac4ed0..94eae97 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.tajo.catalog.CatalogProtocol.CatalogProtocolService;
import org.apache.tajo.catalog.exception.*;
+import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.proto.CatalogProtos.*;
import org.apache.tajo.catalog.store.CatalogStore;
import org.apache.tajo.catalog.store.DBStore;
@@ -173,7 +174,11 @@ public class CatalogServer extends AbstractService {
if (!store.existTable(tableId)) {
throw new NoSuchTableException(tableId);
}
- return (TableDescProto) store.getTable(tableId).getProto();
+ TableDesc desc = store.getTable(tableId);
+ SchemaProto schemaProto = desc.getMeta().getSchema().getProto();
+ SchemaProto qualifiedSchema = CatalogUtil.getQualfiedSchema(tableId, schemaProto);
+ desc.getMeta().setSchema(new Schema(qualifiedSchema));
+ return (TableDescProto) desc.getProto();
} catch (IOException ioe) {
// TODO - handle exception
LOG.error(ioe);
@@ -228,12 +233,8 @@ public class CatalogServer extends AbstractService {
}
// rewrite schema
- SchemaProto revisedSchema =
- CatalogUtil.getQualfiedSchema(tableDesc.getId(), tableDesc.getMeta()
- .getSchema());
-
TableProto.Builder metaBuilder = TableProto.newBuilder(tableDesc.getMeta());
- metaBuilder.setSchema(revisedSchema);
+ metaBuilder.setSchema(tableDesc.getMeta().getSchema());
TableDescProto.Builder descBuilder = TableDescProto.newBuilder(tableDesc);
descBuilder.setMeta(metaBuilder.build());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-common/src/main/java/org/apache/tajo/ExecutionBlockId.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/ExecutionBlockId.java b/tajo-common/src/main/java/org/apache/tajo/ExecutionBlockId.java
index 2dc4441..1ccb357 100644
--- a/tajo-common/src/main/java/org/apache/tajo/ExecutionBlockId.java
+++ b/tajo-common/src/main/java/org/apache/tajo/ExecutionBlockId.java
@@ -32,16 +32,6 @@ public class ExecutionBlockId implements Comparable<ExecutionBlockId> {
this(new QueryId(proto.getQueryId()), proto.getId());
}
-// public ExecutionBlockId(String idStr) {
-// String[] tokens = idStr.split(QueryId.SEPARATOR);
-// if(tokens.length < 3) {
-// throw new RuntimeException("Wrong ExecutionBlockId format[" + idStr + "]");
-// }
-//
-// this.queryId = new QueryId(tokens);
-// this.id = Integer.parseInt(tokens[3]);
-// }
-
@Override
public String toString() {
return EB_ID_PREFIX + QueryId.SEPARATOR + toStringNoPrefix();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-common/src/main/java/org/apache/tajo/QueryIdFactory.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/QueryIdFactory.java b/tajo-common/src/main/java/org/apache/tajo/QueryIdFactory.java
index 90533e3..b0affe9 100644
--- a/tajo-common/src/main/java/org/apache/tajo/QueryIdFactory.java
+++ b/tajo-common/src/main/java/org/apache/tajo/QueryIdFactory.java
@@ -71,14 +71,6 @@ public class QueryIdFactory {
return new QueryId(String.valueOf(timestamp), seq);
}
- /**
- * for test
- * @return
- */
- public synchronized static QueryId newQueryId() {
- return newQueryId(TajoIdUtils.MASTER_ID_FORMAT.format(0));
- }
-
public synchronized static QueryId newQueryId(String seedQueryId, int seq) {
if(isYarnId(seedQueryId)) {
String[] tokens = seedQueryId.split("_");
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java b/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
index 05c0972..373292b 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/TUtil.java
@@ -18,9 +18,6 @@
package org.apache.tajo.util;
-import org.apache.tajo.QueryIdFactory;
-import org.apache.tajo.QueryUnitAttemptId;
-
import java.util.*;
/**
@@ -86,6 +83,22 @@ public class TUtil {
return new HashMap<K, V>(map);
}
+ public static <K, V> Map<K,V> newHashMap(K k, V v) {
+ HashMap<K, V> newMap = new HashMap<K, V>();
+ newMap.put(k, v);
+ return newMap;
+ }
+
+ public static <K,V> Map<K,V> newLinkedHashMap() {
+ return new LinkedHashMap<K, V>();
+ }
+
+ public static <K, V> Map<K,V> newLinkedHashMap(K k, V v) {
+ HashMap<K, V> newMap = new LinkedHashMap<K, V>();
+ newMap.put(k, v);
+ return newMap;
+ }
+
public static <T> List<T> newList() {
return new ArrayList<T>();
}
@@ -108,13 +121,6 @@ public class TUtil {
return list;
}
- public static QueryUnitAttemptId newQueryUnitAttemptId() {
- return QueryIdFactory.newQueryUnitAttemptId(
- QueryIdFactory.newQueryUnitId(
- QueryIdFactory.newExecutionBlockId(
- QueryIdFactory.newQueryId())), 0);
- }
-
/**
* It check if T is null or not.
*
@@ -128,4 +134,13 @@ public class TUtil {
}
return reference;
}
+
+ public static <KEY1, KEY2, VALUE> void putToNestedMap(Map<KEY1, Map<KEY2, VALUE>> map, KEY1 k1, KEY2 k2,
+ VALUE value) {
+ if (map.containsKey(k1)) {
+ map.get(k1).put(k2, value);
+ } else {
+ map.put(k1, TUtil.newLinkedHashMap(k2, value));
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q12.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q12.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q12.tql
index cfdd0f5..2f26ee9 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q12.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q12.tql
@@ -1,7 +1,13 @@
select
l_shipmode,
- sum(case when o_orderpriority ='1-URGENT' or o_orderpriority ='2-HIGH' then 1 else 0 end) as high_line_count,
- sum(case when o_orderpriority != '1-URGENT' and o_orderpriority != '2-HIGH' then 1 else 0 end) as low_line_count
+ sum(case when o_orderpriority ='1-URGENT' or o_orderpriority ='2-HIGH'
+ then 1
+ else 0
+ end) as high_line_count,
+ sum(case when o_orderpriority != '1-URGENT' and o_orderpriority != '2-HIGH'
+ then 1
+ else 0
+ end) as low_line_count
from
orders,
lineitem
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q13.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q13.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q13.tql
index a840a23..2ee0102 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q13.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q13.tql
@@ -1,9 +1,18 @@
select
- c_custkey,
- count(o_orderkey)
-from
- customer
-left outer join
- orders on c_custkey = o_custkey and o_comment not like '%special%requests%'
+ c_count, count(*) as custdist
+from (
+ select
+ c_custkey,
+ count(o_orderkey)
+ from
+ customer left outer join orders on
+ c_custkey = o_custkey
+ and o_comment not like '%special%requests%'
+ group by
+ c_custkey
+ ) as c_orders (c_custkey, c_count)
group by
- c_custkey
\ No newline at end of file
+ c_count
+order by
+ custdist desc,
+ c_count desc
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q14.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q14.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q14.tql
index 7614904..f00e469 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q14.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q14.tql
@@ -1,9 +1,13 @@
select
- 100.00 * sum(case when p_type like 'PROMO%' then l_extendedprice*(1-l_discount) else 0 end) / sum(l_extendedprice * (1 - l_discount)) as promo_revenue
+ 100.00 * sum(case
+ when p_type like 'PROMO%'
+ then l_extendedprice*(1-l_discount)
+ else 0
+ end) / sum(l_extendedprice * (1 - l_discount)) as promo_revenue
from
lineitem,
part
where
l_partkey = p_partkey and
- l_shipdate >= date '1995-09-01' and
- l_shipdate < '1995-10-01'
\ No newline at end of file
+ l_shipdate >= '1995-09-01' and
+ l_shipdate < '1995-10-01';
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q17.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q17.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q17.tql
index bfc7ff6..d67a399 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q17.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q17.tql
@@ -14,4 +14,4 @@ where
lineitem
where
l_partkey = p_partkey
- )
+ )
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q2.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q2.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q2.tql
index c0bf3c2..b7a7cd8 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q2.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q2.tql
@@ -1,24 +1,38 @@
select
- s_acctbal,
- s_name,
- n_name,
- p_partkey,
- p_mfgr,
- s_address,
- s_phone,
- s_comment,
- ps_supplycost
-from
- region
-join
- nation on n_regionkey = r_regionkey
-join
- supplier on s_nationkey = n_nationkey
-join
- partsupp on s_suppkey = ps_suppkey
-join
- part on p_partkey = ps_partkey
+ s_acctbal,
+ s_name,
+ n_name,
+ p_partkey,
+ p_mfgr,
+ s_address,
+ s_phone,
+ s_comment
+from
+ part,
+ supplier,
+ partsupp,
+ nation,
+ region
where
- p_size = 15 and p_type
-like
- '%BRASS' and r_name = 'EUROPE'
\ No newline at end of file
+ p_partkey = ps_partkey
+ and s_suppkey = ps_suppkey
+ and p_size = 15
+ and p_type like '%BRASS'
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and r_name = 'c'
+ and ps_supplycost =
+ (
+ select min(ps_supplycost) from partsupp, supplier, nation, region
+ where
+ p_partkey = ps_partkey
+ and s_suppkey = ps_suppkey
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and r_name = 'EUROPE'
+ )
+order by
+ s_acctbal desc,
+ n_name,
+ s_name,
+ p_partkey
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q4.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q4.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q4.tql
index 760d183..2ab4598 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q4.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q4.tql
@@ -1,7 +1,21 @@
select
- *
-from
- lineitem
-where
- l_orderkey = o_orderkey and
- l_commitdate < l_receiptdate
\ No newline at end of file
+ o_orderpriority,
+ count(*) as order_count
+from
+ orders
+where
+ o_orderdate >= '1993-07-01'
+ and o_orderdate < '1993-10-01'
+ and exists (
+ select
+ *
+ from
+ lineitem
+ where
+ l_orderkey = o_orderkey
+ and l_commitdate < l_receiptdate
+ )
+group by
+ o_orderpriority
+order by
+ o_orderpriority;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q7.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q7.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q7.tql
index 9cf48d6..a7d0f79 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q7.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q7.tql
@@ -1,21 +1,38 @@
select
- n1.n_name as supp_nation,
- n2.n_name as cust_nation,
- extract(year from l_shipdate) as l_year,
- l_extendedprice * (1 - l_discount) as volume
-from
- supplier,
- lineitem,
- orders,
- customer,
- nation n1,
- nation n2
-where
- s_suppkey = l_suppkey and
- o_orderkey = l_orderkey and
- c_custkey = o_custkey and
- s_nationkey = n1.n_nationkey and
- c_nationkey = n2.n_nationkey and
- ( (n1.n_name = 'FRANCE' and n2.n_name = 'GERMANY') or (n1.n_name = 'GERMANY' and n2.n_name = 'FRANCE') ) and
- l_shipdate > '1995-01-01' and
- l_shipdate < '1996-12-31' as shipping
\ No newline at end of file
+ supp_nation,
+ cust_nation,
+ l_year,
+ sum(volume) as revenue
+from (
+ select
+ n1.n_name as supp_nation,
+ n2.n_name as cust_nation,
+ l_shipdate as l_year,
+ l_extendedprice * (1 - l_discount) as volume
+ from
+ supplier,
+ lineitem,
+ orders,
+ customer,
+ nation n1,
+ nation n2
+ where
+ s_suppkey = l_suppkey
+ and o_orderkey = l_orderkey
+ and c_custkey = o_custkey
+ and s_nationkey = n1.n_nationkey
+ and c_nationkey = n2.n_nationkey
+ and (
+ (n1.n_name = 'FRANCE' and n2.n_name = 'GERMANY')
+ or (n1.n_name = 'GERMANY' and n2.n_name = 'FRANCE')
+ )
+ and '1995-01-01' < l_shipdate and l_shipdate < '1996-12-31'
+ ) as shipping
+group by
+ supp_nation,
+ cust_nation,
+ l_year
+order by
+ supp_nation,
+ cust_nation,
+ l_year;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q8.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q8.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q8.tql
index 3dbde76..9e96d4d 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q8.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q8.tql
@@ -1,25 +1,37 @@
select
- extract(year from o_orderdate) as o_year,
- l_extendedprice * (1-l_discount) as volume,
- n2.n_name as nation
-from
- part,
- supplier,
- lineitem,
- orders,
- customer,
- nation n1,
- nation n2,
- region
-where
- p_partkey = l_partkey and
- s_suppkey = l_suppkey and
- l_orderkey = o_orderkey and
- o_custkey = c_custkey and
- c_nationkey = n1.n_nationkey and
- n1.n_regionkey = r_regionkey and
- r_name = 'AMERICA' and
- s_nationkey = n2.n_nationkey and
- o_orderdate > '1995-01-01' and
- o_orderdate < '1996-12-31' and
- p_type = 'ECONOMY ANODIZED STEEL' as all_nations
\ No newline at end of file
+ o_year,
+ sum(case
+ when nation = 'BRAZIL'
+ then volume
+ else 0
+ end) / sum(volume) as mkt_share
+from (
+ select
+ o_orderdate as o_year,
+ l_extendedprice * (1-l_discount) as volume,
+ n2.n_name as nation
+ from
+ part,
+ supplier,
+ lineitem,
+ orders,
+ customer,
+ nation n1,
+ nation n2,
+ region
+ where
+ p_partkey = l_partkey
+ and s_suppkey = l_suppkey
+ and l_orderkey = o_orderkey
+ and o_custkey = c_custkey
+ and c_nationkey = n1.n_nationkey
+ and n1.n_regionkey = r_regionkey
+ and r_name = 'AMERICA'
+ and s_nationkey = n2.n_nationkey
+ and '1995-01-01' < o_orderdate and o_orderdate <'1996-12-31'
+ and p_type = 'ECONOMY ANODIZED STEEL'
+ ) as all_nations
+group by
+ o_year
+order by
+ o_year;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/benchmark/tpch/q9.tql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/benchmark/tpch/q9.tql b/tajo-core/tajo-core-backend/benchmark/tpch/q9.tql
index 396c9c3..c22523e 100644
--- a/tajo-core/tajo-core-backend/benchmark/tpch/q9.tql
+++ b/tajo-core/tajo-core-backend/benchmark/tpch/q9.tql
@@ -1,19 +1,31 @@
-select
- n_name as nation,
- extract(year from o_orderdate) as o_year,
- l_extendedprice * (1 - l_discount) - ps_supplycost * l_quantity as amount
-from
- part,
- supplier,
- lineitem,
- partsupp,
- orders,
- nation
-where
- s_suppkey = l_suppkey and
- ps_suppkey = l_suppkey and
- ps_partkey = l_partkey and
- p_partkey = l_partkey and
- o_orderkey = l_orderkey and
- s_nationkey = n_nationkey and
- p_name like '%green%' ) as profit
\ No newline at end of file
+select
+ nation,
+ o_year,
+ sum(amount) as sum_profit
+ from (
+ select
+ n_name as nation,
+ o_orderdate as o_year,
+ l_extendedprice * (1 - l_discount) - ps_supplycost * l_quantity as amount
+ from
+ part,
+ supplier,
+ lineitem,
+ partsupp,
+ orders,
+ nation
+ where
+ s_suppkey = l_suppkey
+ and ps_suppkey = l_suppkey
+ and ps_partkey = l_partkey
+ and p_partkey = l_partkey
+ and o_orderkey = l_orderkey
+ and s_nationkey = n_nationkey
+ and p_name like '%[COLOR]%'
+ ) as profit
+group by
+ nation,
+ o_year
+order by
+ nation,
+ o_year desc;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/DataChannel.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/DataChannel.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/DataChannel.java
new file mode 100644
index 0000000..be1852d
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/DataChannel.java
@@ -0,0 +1,198 @@
+/**
+ * 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;
+
+import com.google.common.base.Preconditions;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.master.ExecutionBlock;
+
+import static org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
+import static org.apache.tajo.ipc.TajoWorkerProtocol.*;
+
+public class DataChannel {
+ private ExecutionBlockId srcId;
+ private ExecutionBlockId targetId;
+ private TransmitType transmitType = TransmitType.PULL_TRANSMIT;
+ private PartitionType partitionType;
+ private Integer partitionNum = 1;
+ private Column[] key;
+
+ private Schema schema;
+
+ private StoreType storeType = StoreType.CSV;
+
+ public DataChannel(ExecutionBlockId srcId, ExecutionBlockId targetId) {
+ this.srcId = srcId;
+ this.targetId = targetId;
+ }
+
+ public DataChannel(ExecutionBlockId srcId, ExecutionBlockId targetId, PartitionType partitionType) {
+ this(srcId, targetId);
+ this.partitionType = partitionType;
+ }
+
+ public DataChannel(ExecutionBlock src, ExecutionBlock target, PartitionType partitionType, int partNum) {
+ this(src.getId(), target.getId(), partitionType, partNum);
+ setSchema(src.getPlan().getOutSchema());
+ }
+
+ public DataChannel(ExecutionBlockId srcId, ExecutionBlockId targetId, PartitionType partitionType, int partNum) {
+ this(srcId, targetId, partitionType);
+ this.partitionNum = partNum;
+ }
+
+ public DataChannel(DataChannelProto proto) {
+ this.srcId = new ExecutionBlockId(proto.getSrcId());
+ this.targetId = new ExecutionBlockId(proto.getTargetId());
+ this.transmitType = proto.getTransmitType();
+ this.partitionType = proto.getPartitionType();
+ if (proto.hasSchema()) {
+ this.setSchema(new Schema(proto.getSchema()));
+ }
+ if (proto.getPartitionKeyCount() > 0) {
+ key = new Column[proto.getPartitionKeyCount()];
+ for (int i = 0; i < proto.getPartitionKeyCount(); i++) {
+ key[i] = new Column(proto.getPartitionKey(i));
+ }
+ } else {
+ key = new Column[] {};
+ }
+ if (proto.hasPartitionNum()) {
+ this.partitionNum = proto.getPartitionNum();
+ }
+ }
+
+ public ExecutionBlockId getSrcId() {
+ return srcId;
+ }
+
+ public ExecutionBlockId getTargetId() {
+ return targetId;
+ }
+
+ public PartitionType getPartitionType() {
+ return partitionType;
+ }
+
+ public TransmitType getTransmitType() {
+ return this.transmitType;
+ }
+
+ public void setTransmitType(TransmitType transmitType) {
+ this.transmitType = transmitType;
+ }
+
+ public void setPartition(PartitionType partitionType, Column [] keys, int numPartitions) {
+ Preconditions.checkArgument(keys.length >= 0, "At least one partition key must be specified.");
+ Preconditions.checkArgument(numPartitions > 0, "The number of partitions must be positive: %s", numPartitions);
+
+ this.partitionType = partitionType;
+ this.key = keys;
+ this.partitionNum = numPartitions;
+ }
+
+ public void setPartitionType(PartitionType partitionType) {
+ this.partitionType = partitionType;
+ }
+
+ public boolean hasPartitionKey() {
+ return key != null;
+ }
+
+ public void setPartitionKey(Column [] key) {
+ this.key = key;
+ }
+
+ public Column [] getPartitionKey() {
+ return this.key;
+ }
+
+ public void setPartitionNum(int partNum) {
+ this.partitionNum = partNum;
+ }
+
+ public int getPartitionNum() {
+ return partitionNum;
+ }
+
+ public boolean hasStoreType() {
+ return this.storeType != null;
+ }
+
+ public void setStoreType(StoreType storeType) {
+ this.storeType = storeType;
+ }
+
+ public StoreType getStoreType() {
+ return storeType;
+ }
+
+ public DataChannelProto getProto() {
+ DataChannelProto.Builder builder = DataChannelProto.newBuilder();
+ builder.setSrcId(srcId.getProto());
+ builder.setTargetId(targetId.getProto());
+ if (transmitType != null) {
+ builder.setTransmitType(transmitType);
+ }
+ builder.setPartitionType(partitionType);
+ if (schema != null) {
+ builder.setSchema(schema.getProto());
+ }
+ if (key != null) {
+ for (Column column : key) {
+ builder.addPartitionKey(column.getProto());
+ }
+ }
+ if (partitionNum != null) {
+ builder.setPartitionNum(partitionNum);
+ }
+ return builder.build();
+ }
+
+ public void setSchema(Schema schema) {
+ this.schema = schema;
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("[").append(srcId.getQueryId()).append("] ");
+ sb.append(srcId.getId()).append(" => ").append(targetId.getId());
+ sb.append(" (type=").append(partitionType);
+ if (hasPartitionKey()) {
+ sb.append(", key=");
+ boolean first = true;
+ for (Column column : getPartitionKey()) {
+ if (first) {
+ first = false;
+ } else {
+ sb.append(",");
+ }
+ sb.append(column.getColumnName());
+ }
+ sb.append(", num=").append(partitionNum);
+ }
+ sb.append(")");
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/TaskAttemptContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/TaskAttemptContext.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/TaskAttemptContext.java
index 8ae2f2b..7eac43d 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/TaskAttemptContext.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/TaskAttemptContext.java
@@ -54,6 +54,7 @@ public class TaskAttemptContext {
private boolean stopped = false;
private boolean interQuery = false;
private Path outputPath;
+ private DataChannel dataChannel;
public TaskAttemptContext(TajoConf conf, final QueryUnitAttemptId queryId,
final Fragment[] fragments,
@@ -90,6 +91,14 @@ public class TaskAttemptContext {
LOG.info("Query status of " + getTaskId() + " is changed to " + state);
}
+ public void setDataChannel(DataChannel dataChannel) {
+ this.dataChannel = dataChannel;
+ }
+
+ public DataChannel getDataChannel() {
+ return dataChannel;
+ }
+
public boolean hasResultStats() {
return resultStats != null;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/EvalTreeUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/EvalTreeUtil.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/EvalTreeUtil.java
index 3fc41dd..e6bd733 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/EvalTreeUtil.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/EvalTreeUtil.java
@@ -31,13 +31,7 @@ import org.apache.tajo.exception.InternalException;
import java.util.*;
public class EvalTreeUtil {
- public static void changeColumnRef(EvalNode node, Column oldName,
- Column newName) {
- node.postOrder(new ChangeColumnRefVisitor(oldName.getQualifiedName(),
- newName.getQualifiedName()));
- }
-
- public static void changeColumnRef(EvalNode node, String oldName,
+ public static void changeColumnRef(EvalNode node, String oldName,
String newName) {
node.postOrder(new ChangeColumnRefVisitor(oldName, newName));
}
@@ -131,7 +125,7 @@ public class EvalTreeUtil {
case FIELD:
FieldEval fieldEval = (FieldEval) expr;
- return SchemaUtil.newNoNameSchema(inputSchema.getColumn(fieldEval.getName()).getDataType());
+ return SchemaUtil.newNoNameSchema(inputSchema.getColumnByFQN(fieldEval.getName()).getDataType());
default:
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/FieldEval.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/FieldEval.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/FieldEval.java
index cd9c48d..87d6077 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/FieldEval.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/FieldEval.java
@@ -47,7 +47,7 @@ public class FieldEval extends EvalNode implements Cloneable {
fieldId = schema.getColumnId(column.getQualifiedName());
} else {
if(schema.getColumnNum() != 0) {
- String schemaColQualName = schema.getColumn(0).getTableName() +
+ String schemaColQualName = schema.getColumn(0).getQualifier() +
"." + column.getColumnName();
fieldId = schema.getColumnId(schemaColQualName);
} else {
@@ -86,7 +86,7 @@ public class FieldEval extends EvalNode implements Cloneable {
}
public String getTableId() {
- return column.getTableName();
+ return column.getQualifier();
}
public String getColumnName() {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
index 96e67f2..133fc50 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
@@ -21,7 +21,6 @@ package org.apache.tajo.engine.parser;
import com.google.common.base.Preconditions;
import org.antlr.v4.runtime.ANTLRInputStream;
import org.antlr.v4.runtime.CommonTokenStream;
-import org.antlr.v4.runtime.misc.NotNull;
import org.antlr.v4.runtime.tree.TerminalNode;
import org.apache.tajo.algebra.*;
import org.apache.tajo.algebra.Aggregation.GroupType;
@@ -82,7 +81,7 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
OpType operatorType;
Expr left;
- for (int i = 1; i < ctx.getChildCount();) {
+ for (int i = 1; i < ctx.getChildCount(); i++) {
int idx = i;
boolean distinct = true;
@@ -110,7 +109,7 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
left = current;
current = new SetOperation(operatorType, left, right, distinct);
- i+=idx;
+ i = idx;
}
}
@@ -254,7 +253,7 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
public RelationList visitFrom_clause(SQLParser.From_clauseContext ctx) {
Expr [] relations = new Expr[ctx.table_reference_list().table_reference().size()];
for (int i = 0; i < relations.length; i++) {
- relations[i] = visit(ctx.table_reference_list().table_reference(i));
+ relations[i] = visitTable_reference(ctx.table_reference_list().table_reference(i));
}
return new RelationList(relations);
}
@@ -391,7 +390,7 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
public Expr visitTable_primary(SQLParser.Table_primaryContext ctx) {
if (ctx.table_or_query_name() != null) {
Relation relation = new Relation(ctx.table_or_query_name().getText());
- if (ctx.AS() != null) {
+ if (ctx.alias != null) {
relation.setAlias(ctx.alias.getText());
}
return relation;
@@ -656,7 +655,7 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
}
}
- @Override public FunctionExpr visitAggregate_function(@NotNull SQLParser.Aggregate_functionContext ctx) {
+ @Override public FunctionExpr visitAggregate_function( SQLParser.Aggregate_functionContext ctx) {
if (ctx.COUNT() != null && ctx.MULTIPLY() != null) {
return new CountRowsFunctionExpr();
} else {
@@ -664,7 +663,7 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
}
}
- @Override public FunctionExpr visitGeneral_set_function(@NotNull SQLParser.General_set_functionContext ctx) {
+ @Override public FunctionExpr visitGeneral_set_function(SQLParser.General_set_functionContext ctx) {
String signature = ctx.set_function_type().getText();
boolean distinct = checkIfExist(ctx.set_qualifier()) && checkIfExist(ctx.set_qualifier().DISTINCT()) ? true : false;
Expr param = visitBoolean_value_expression(ctx.boolean_value_expression());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java
index 9dba5e1..78422b0 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java
@@ -33,6 +33,7 @@ public interface AlgebraVisitor<T1, T2> {
T2 visitExcept(T1 ctx, Stack<OpType> stack, SetOperation expr) throws PlanningException;
T2 visitIntersect(T1 ctx, Stack<OpType> stack, SetOperation expr) throws PlanningException;
T2 visitRelationList(T1 ctx, Stack<OpType> stack, RelationList expr) throws PlanningException;
+ T2 visitTableSubQuery(T1 ctx, Stack<OpType> stack, TableSubQuery expr) throws PlanningException;
T2 visitRelation(T1 ctx, Stack<OpType> stack, Relation expr) throws PlanningException;
T2 visitCreateTable(T1 ctx, Stack<OpType> stack, CreateTable expr) throws PlanningException;
T2 visitDropTable(T1 ctx, Stack<OpType> stack, DropTable expr) throws PlanningException;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
index 4c06f28..bac146c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
@@ -79,6 +79,9 @@ public abstract class BaseAlgebraVisitor<T1, T2> implements AlgebraVisitor<T1, T
case RelationList:
current = visitRelationList(ctx, stack, (RelationList) expr);
break;
+ case TableSubQuery:
+ current = visitTableSubQuery(ctx, stack, (TableSubQuery) expr);
+ break;
case Relation:
current = visitRelation(ctx, stack, (Relation) expr);
break;
@@ -195,6 +198,14 @@ public abstract class BaseAlgebraVisitor<T1, T2> implements AlgebraVisitor<T1, T
}
@Override
+ public T2 visitTableSubQuery(T1 ctx, Stack<OpType> stack, TableSubQuery expr) throws PlanningException {
+ stack.push(expr.getType());
+ T2 child = visitChild(ctx, stack, expr.getSubQuery());
+ stack.pop();
+ return child;
+ }
+
+ @Override
public T2 visitRelation(T1 ctx, Stack<OpType> stack, Relation expr) throws PlanningException {
return null;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
index 52d2a0a..b58b05e 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
@@ -76,6 +76,9 @@ public class BasicLogicalPlanVisitor<T> implements LogicalPlanVisitor<T> {
case INTERSECT:
current = visitIntersect(plan, (IntersectNode) node, stack, data);
break;
+ case TABLE_SUBQUERY:
+ current = visitTableSubQuery(plan, (TableSubQueryNode) node, stack, data);
+ break;
case SCAN:
current = visitScan(plan, (ScanNode) node, stack, data);
break;
@@ -184,6 +187,15 @@ public class BasicLogicalPlanVisitor<T> implements LogicalPlanVisitor<T> {
}
@Override
+ public LogicalNode visitTableSubQuery(LogicalPlan plan, TableSubQueryNode node, Stack<LogicalNode> stack, T data)
+ throws PlanningException {
+ stack.push(node);
+ visitChild(plan, node.getSubQuery(), stack, data);
+ stack.pop();
+ return node;
+ }
+
+ @Override
public LogicalNode visitScan(LogicalPlan plan, ScanNode node, Stack<LogicalNode> stack, T data)
throws PlanningException {
return node;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java
new file mode 100644
index 0000000..d572494
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java
@@ -0,0 +1,202 @@
+/**
+ * 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;
+
+import org.apache.tajo.engine.planner.logical.*;
+
+import java.util.Stack;
+
+/**
+ * It returns a list of node plan strings.
+ */
+public class ExplainLogicalPlanVisitor extends BasicLogicalPlanVisitor<ExplainLogicalPlanVisitor.Context> {
+
+ public static class Context {
+ public int maxDepth = -1;
+ public int depth = 0;
+ public Stack<DepthString> explains = new Stack<DepthString>();
+
+ public void add(int depth, PlanString planString) {
+ maxDepth = Math.max(maxDepth, depth);
+ explains.push(new DepthString(depth, planString));
+ }
+
+ public int getMaxDepth() {
+ return this.maxDepth;
+ }
+
+ public Stack<DepthString> getExplains() {
+ return explains;
+ }
+ }
+
+ public static class DepthString {
+ private int depth;
+ private PlanString planStr;
+
+ DepthString(int depth, PlanString planStr) {
+ this.depth = depth;
+ this.planStr = planStr;
+ }
+
+ public int getDepth() {
+ return depth;
+ }
+
+ public PlanString getPlanString() {
+ return planStr;
+ }
+ }
+
+ public Context getBlockPlanStrings(LogicalPlan plan, String block) throws PlanningException {
+ Stack<LogicalNode> stack = new Stack<LogicalNode>();
+ Context explainContext = new Context();
+ visitChild(plan, plan.getBlock(block).getRoot(), stack, explainContext);
+ return explainContext;
+ }
+
+ @Override
+ public LogicalNode visitRoot(LogicalPlan plan, LogicalRootNode node, Stack<LogicalNode> stack, Context data)
+ throws PlanningException {
+ return visitChild(plan, node.getChild(), stack, data);
+ }
+
+ public LogicalNode visitProjection(LogicalPlan plan, ProjectionNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ return visitUnaryNode(plan, node, stack, context);
+ }
+
+ @Override
+ public LogicalNode visitLimit(LogicalPlan plan, LimitNode node, Stack<LogicalNode> stack, Context context)
+ throws PlanningException {
+ return visitUnaryNode(plan, node, stack, context);
+ }
+
+ public LogicalNode visitSort(LogicalPlan plan, SortNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ return visitUnaryNode(plan, node, stack, context);
+ }
+
+ public LogicalNode visitGroupBy(LogicalPlan plan, GroupbyNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ return visitUnaryNode(plan, node, stack, context);
+ }
+
+ private LogicalNode visitUnaryNode(LogicalPlan plan, UnaryNode node, Stack<LogicalNode> stack, Context context)
+ throws PlanningException {
+ context.depth++;
+ stack.push(node);
+ visitChild(plan, node.getChild(), stack, context);
+ context.depth--;
+ context.add(context.depth, node.getPlanString());
+ return node;
+ }
+
+ private LogicalNode visitBinaryNode(LogicalPlan plan, BinaryNode node, Stack<LogicalNode> stack, Context context)
+ throws PlanningException {
+ context.depth++;
+ stack.push(node);
+ visitChild(plan, node.getLeftChild(), stack, context);
+ visitChild(plan, node.getRightChild(), stack, context);
+ stack.pop();
+ context.depth--;
+ context.add(context.depth, node.getPlanString());
+ return node;
+ }
+
+ public LogicalNode visitFilter(LogicalPlan plan, SelectionNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ return visitUnaryNode(plan, node, stack, context);
+ }
+
+ @Override
+ public LogicalNode visitJoin(LogicalPlan plan, JoinNode node, Stack<LogicalNode> stack, Context context)
+ throws PlanningException {
+ return visitBinaryNode(plan, node, stack, context);
+ }
+
+ @Override
+ public LogicalNode visitUnion(LogicalPlan plan, UnionNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ return visitBinaryNode(plan, node, stack, context);
+ }
+
+ @Override
+ public LogicalNode visitExcept(LogicalPlan plan, ExceptNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ return visitBinaryNode(plan, node, stack, context);
+ }
+
+ @Override
+ public LogicalNode visitIntersect(LogicalPlan plan, IntersectNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ return visitBinaryNode(plan, node, stack, context);
+ }
+
+ @Override
+ public LogicalNode visitTableSubQuery(LogicalPlan plan, TableSubQueryNode node, Stack<LogicalNode> stack,
+ Context context) throws PlanningException {
+ context.depth++;
+ stack.push(node);
+ super.visitTableSubQuery(plan, node, stack, context);
+ stack.pop();
+ context.depth--;
+ context.add(context.depth, node.getPlanString());
+
+ return node;
+ }
+
+ @Override
+ public LogicalNode visitScan(LogicalPlan plan, ScanNode node, Stack<LogicalNode> stack, Context context)
+ throws PlanningException {
+ context.add(context.depth, node.getPlanString());
+ return node;
+ }
+
+ @Override
+ public LogicalNode visitStoreTable(LogicalPlan plan, StoreTableNode node, Stack<LogicalNode> stack, Context context) throws PlanningException {
+ return visitUnaryNode(plan, node, stack, context);
+ }
+
+ @Override
+ public LogicalNode visitInsert(LogicalPlan plan, InsertNode node, Stack<LogicalNode> stack, Context context) throws PlanningException {
+ context.depth++;
+ stack.push(node);
+ visitChild(plan, node.getSubQuery(), stack, context);
+ stack.pop();
+ context.depth--;
+ context.add(context.depth, node.getPlanString());
+ return node;
+ }
+
+ public static String printDepthString(int maxDepth, DepthString planStr) {
+ StringBuilder output = new StringBuilder();
+ String pad = new String(new char[planStr.getDepth() * 3]).replace('\0', ' ');
+ output.append(pad + planStr.getPlanString().getTitle()).append("\n");
+
+ for (String str : planStr.getPlanString().getExplanations()) {
+ output.append(pad).append(" => ").append(str).append("\n");
+ }
+
+ for (String str : planStr.getPlanString().getDetails()) {
+ output.append(pad).append(" => ").append(str).append("\n");
+ }
+ return output.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java
index f85fae5..65c0dfa 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java
@@ -180,4 +180,16 @@ public class InsertNode extends LogicalNode implements Cloneable {
public void postOrder(LogicalNodeVisitor visitor) {
visitor.visit(this);
}
+
+ @Override
+ public PlanString getPlanString() {
+ PlanString planString = new PlanString("INSERT");
+ planString.addExplan(" INTO ");
+ if (hasTargetTable()) {
+ planString.addExplan(getTargetTable().getName());
+ } else {
+ planString.addExplan("LOCATION " + path);
+ }
+ return planString;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
index ed88cbc..91f3e7c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
@@ -27,6 +27,8 @@ import org.apache.tajo.catalog.Schema;
import org.apache.tajo.engine.eval.EvalNode;
import org.apache.tajo.engine.eval.EvalTreeUtil;
import org.apache.tajo.engine.eval.EvalType;
+import org.apache.tajo.engine.planner.graph.DirectedGraphVisitor;
+import org.apache.tajo.engine.planner.graph.SimpleDirectedGraph;
import org.apache.tajo.engine.planner.logical.*;
import java.util.*;
@@ -52,7 +54,7 @@ public class LogicalPlan {
/** a map from between a block name to a block plan */
private Map<String, QueryBlock> queryBlocks = new LinkedHashMap<String, QueryBlock>();
private Map<LogicalNode, QueryBlock> queryBlockByNode = new HashMap<LogicalNode, QueryBlock>();
- private QueryBlockGraph queryBlockGraph = new QueryBlockGraph();
+ private SimpleDirectedGraph<QueryBlock, BlockEdge> blockGraph = new SimpleDirectedGraph<QueryBlock, BlockEdge>();
private Set<LogicalNode> visited = new HashSet<LogicalNode>();
public LogicalPlan(LogicalPlanner planner) {
@@ -111,18 +113,22 @@ public class LogicalPlan {
}
}
- public void connectBlocks(QueryBlock srcBlock, QueryBlock targetBlock, QueryBlockGraph.BlockType type) {
- queryBlockGraph.connectBlocks(srcBlock.getName(), targetBlock.getName(), type);
+ public void connectBlocks(QueryBlock srcBlock, QueryBlock targetBlock, BlockType type) {
+ blockGraph.connect(srcBlock, targetBlock, new BlockEdge(srcBlock, targetBlock, type));
}
- public Collection<QueryBlockGraph.BlockEdge> getConnectedBlocks(String blockName) {
- return queryBlockGraph.getBlockEdges(blockName);
+ public List<QueryBlock> getChildBlocks(QueryBlock block) {
+ return blockGraph.getChilds(block);
}
public Collection<QueryBlock> getQueryBlocks() {
return queryBlocks.values();
}
+ public SimpleDirectedGraph<QueryBlock, BlockEdge> getBlockGraph() {
+ return blockGraph;
+ }
+
public boolean postVisit(String blockName, LogicalNode node, Stack<OpType> path) {
if (visited.contains(node)) {
return false;
@@ -142,7 +148,7 @@ public class LogicalPlan {
break;
case GROUP_BY:
- block.resolveGrouping();
+ block.needToResolveGrouping();
break;
case SELECTION:
@@ -152,6 +158,11 @@ public class LogicalPlan {
case INSERT:
block.setInsertNode((InsertNode) node);
break;
+
+ case TABLE_SUBQUERY:
+ TableSubQueryNode tableSubQueryNode = (TableSubQueryNode) node;
+ block.addRelation(tableSubQueryNode);
+ break;
}
@@ -172,7 +183,7 @@ public class LogicalPlan {
throws VerifyException {
QueryBlock block = queryBlocks.get(blockName);
- ScanNode relationOp = block.getRelation(relName);
+ RelationNode relationOp = block.getRelation(relName);
// if a column name is outside of this query block
if (relationOp == null) {
@@ -200,7 +211,7 @@ public class LogicalPlan {
} catch (CloneNotSupportedException e) {
e.printStackTrace();
}
- String tableName = relationOp.getTableId();
+ String tableName = relationOp.getTableName();
column.setName(tableName + "." + column.getColumnName());
return column;
@@ -218,11 +229,11 @@ public class LogicalPlan {
Column candidate;
if (columnRef.hasTableName()) {
- candidate = node.getOutSchema().getColumn(columnRef.getCanonicalName());
+ candidate = node.getOutSchema().getColumnByFQN(columnRef.getCanonicalName());
if (candidate == null) { // If not found, try to find the column with alias name
- String tableName = getBlock(blockName).getRelation(columnRef.getTableName()).getTableId();
- candidate = node.getOutSchema().getColumn(tableName + "." + columnRef.getName());
+ String tableName = getBlock(blockName).getRelation(columnRef.getTableName()).getTableName();
+ candidate = node.getOutSchema().getColumnByFQN(tableName + "." + columnRef.getName());
}
candidates.add(candidate);
@@ -266,7 +277,7 @@ public class LogicalPlan {
Column candidate;
// Try to find a column from the current query block
- for (ScanNode rel : queryBlocks.get(blockName).getRelations()) {
+ for (RelationNode rel : queryBlocks.get(blockName).getRelations()) {
candidate = findColumnFromRelationOp(rel, name);
if (candidate != null) {
@@ -276,7 +287,7 @@ public class LogicalPlan {
} catch (CloneNotSupportedException e) {
e.printStackTrace();
}
- candidate.setName(rel.getTableId() + "." + candidate.getColumnName());
+ candidate.setName(rel.getTableName() + "." + candidate.getColumnName());
}
candidates.add(candidate);
if (candidates.size() > 0) {
@@ -290,7 +301,7 @@ public class LogicalPlan {
// for each block
Outer:
for (QueryBlock block : queryBlocks.values()) {
- for (ScanNode rel : block.getRelations()) {
+ for (RelationNode rel : block.getRelations()) {
candidate = findColumnFromRelationOp(rel, name);
if (candidate != null) {
@@ -300,7 +311,7 @@ public class LogicalPlan {
} catch (CloneNotSupportedException e) {
e.printStackTrace();
}
- candidate.setName(rel.getTableId() + "." + candidate.getColumnName());
+ candidate.setName(rel.getTableName() + "." + candidate.getColumnName());
}
candidates.add(candidate);
if (candidates.size() > 0)
@@ -311,7 +322,7 @@ public class LogicalPlan {
}
if (candidates.isEmpty()) {
- throw new VerifyException("ERROR: no such a column name "+ name);
+ throw new VerifyException("ERROR: no such a column '"+ name + "'");
} else if (candidates.size() > 1) {
throw new VerifyException("ERROR: column name "+ name + " is ambiguous");
}
@@ -319,7 +330,7 @@ public class LogicalPlan {
return candidates.get(0);
}
- private Column findColumnFromRelationOp(ScanNode relation, String name) throws VerifyException {
+ private Column findColumnFromRelationOp(RelationNode relation, String name) throws VerifyException {
Column candidate = relation.getTableSchema().getColumnByName(name);
if (candidate != null) {
try {
@@ -328,7 +339,7 @@ public class LogicalPlan {
throw new RuntimeException(e);
}
if (!isVirtualRelation(relation.getCanonicalName())) {
- candidate.setName(relation.getTableId() + "." + name);
+ candidate.setName(relation.getTableName() + "." + name);
}
return candidate;
@@ -345,7 +356,7 @@ public class LogicalPlan {
private String blockName;
private LogicalNode rootNode;
private NodeType rootType;
- private Map<String, ScanNode> relations = new HashMap<String, ScanNode>();
+ private Map<String, RelationNode> relations = new HashMap<String, RelationNode>();
private Projection projection;
private boolean resolvedGrouping = true;
@@ -384,8 +395,8 @@ public class LogicalPlan {
queryBlockByNode.put(blockRoot, this);
}
- public LogicalNode getRoot() {
- return rootNode;
+ public <T extends LogicalNode> T getRoot() {
+ return (T) rootNode;
}
public NodeType getRootType() {
@@ -396,15 +407,15 @@ public class LogicalPlan {
return relations.containsKey(name);
}
- public void addRelation(ScanNode relation) {
+ public void addRelation(RelationNode relation) {
relations.put(relation.getCanonicalName(), relation);
}
- public ScanNode getRelation(String name) {
+ public RelationNode getRelation(String name) {
return relations.get(name);
}
- public Collection<ScanNode> getRelations() {
+ public Collection<RelationNode> getRelations() {
return this.relations.values();
}
@@ -440,7 +451,7 @@ public class LogicalPlan {
return this.resolvedGrouping;
}
- public void resolveGrouping() {
+ public void needToResolveGrouping() {
this.resolvedGrouping = true;
}
@@ -524,7 +535,7 @@ public class LogicalPlan {
}
// If all columns are projected and do not include any expression
- if (projection.isAllProjected() && node instanceof ScanNode) {
+ if (projection.isAllProjected() && node instanceof RelationNode) {
targetListManager = new TargetListManager(LogicalPlan.this, PlannerUtil.schemaToTargets(node.getInSchema()));
targetListManager.setEvaluatedAll();
@@ -536,7 +547,7 @@ public class LogicalPlan {
for (int i = 0; i < targetListManager.size(); i++) {
if (targetListManager.getTarget(i) == null) {
try {
- targetListManager.updateTarget(i,planner.createTarget(LogicalPlan.this, blockName,
+ targetListManager.updateTarget(i, planner.createTarget(LogicalPlan.this, blockName,
projection.getTargets()[i]));
} catch (VerifyException e) {
}
@@ -552,7 +563,7 @@ public class LogicalPlan {
if (canTargetEvaluated(i, node)) {
- if (node instanceof ScanNode) { // for scan node
+ if (node instanceof RelationNode) { // for scan node
if (expr.getType() == EvalType.FIELD) {
targetListManager.setEvaluated(i);
if (targetListManager.getTarget(i).hasAlias()) {
@@ -631,4 +642,76 @@ public class LogicalPlan {
return targetListManager.getTargets();
}
}
+
+ public String getQueryGraphAsString() {
+ StringBuilder sb = new StringBuilder();
+
+ sb.append("-----------------------------\n");
+ sb.append("Query Block Graph\n");
+ sb.append("-----------------------------\n");
+ sb.append(blockGraph.toStringGraph(getRootBlock()));
+ sb.append("-----------------------------\n");
+
+ sb.append("\n");
+
+ sb.append(getLogicalPlanAsString());
+
+ return sb.toString();
+ }
+
+ public String getLogicalPlanAsString() {
+ ExplainLogicalPlanVisitor explain = new ExplainLogicalPlanVisitor();
+
+ StringBuilder explains = new StringBuilder();
+ try {
+ ExplainLogicalPlanVisitor.Context explainContext = explain.getBlockPlanStrings(this, ROOT_BLOCK);
+ while(!explainContext.explains.empty()) {
+ explains.append(
+ ExplainLogicalPlanVisitor.printDepthString(explainContext.getMaxDepth(), explainContext.explains.pop()));
+ }
+ } catch (PlanningException e) {
+ e.printStackTrace();
+ }
+
+ return explains.toString();
+ }
+
+ @Override
+ public String toString() {
+ return getQueryGraphAsString();
+ }
+
+ public static enum BlockType {
+ TableSubQuery,
+ ScalarSubQuery
+ }
+
+ public static class BlockEdge {
+ private String childName;
+ private String parentName;
+ private BlockType blockType;
+
+
+ public BlockEdge(String childName, String parentName, BlockType blockType) {
+ this.childName = childName;
+ this.parentName = parentName;
+ this.blockType = blockType;
+ }
+
+ public BlockEdge(QueryBlock child, QueryBlock parent, BlockType blockType) {
+ this(child.getName(), parent.getName(), blockType);
+ }
+
+ public String getParentName() {
+ return parentName;
+ }
+
+ public String getChildName() {
+ return childName;
+ }
+
+ public BlockType getBlockType() {
+ return blockType;
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/1b1d1e8c/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
index c41c534..ca5a8f0 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
@@ -43,6 +43,8 @@ public interface LogicalPlanVisitor <T> {
throws PlanningException;
LogicalNode visitIntersect(LogicalPlan plan, IntersectNode node, Stack<LogicalNode> stack, T data)
throws PlanningException;
+ LogicalNode visitTableSubQuery(LogicalPlan plan, TableSubQueryNode node, Stack<LogicalNode> stack, T data)
+ throws PlanningException;
LogicalNode visitScan(LogicalPlan plan, ScanNode node, Stack<LogicalNode> stack, T data)
throws PlanningException;
LogicalNode visitStoreTable(LogicalPlan plan, StoreTableNode node, Stack<LogicalNode> stack, T data)