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/01/07 02:26:42 UTC
git commit: TAJO-479: Rename obsolete name 'partition' to 'shuffle
and fix the broken taskdetail.jsp.
Updated Branches:
refs/heads/master a65f8a10a -> f825dcebc
TAJO-479: Rename obsolete name 'partition' to 'shuffle and fix the broken taskdetail.jsp.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/f825dceb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/f825dceb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/f825dceb
Branch: refs/heads/master
Commit: f825dcebca7ab625256afaf700c103e21e804b35
Parents: a65f8a1
Author: Hyunsik Choi <hy...@apache.org>
Authored: Tue Jan 7 10:26:29 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Tue Jan 7 10:26:29 2014 +0900
----------------------------------------------------------------------
CHANGES.txt | 3 +
.../tajo/catalog/statistics/StatisticsUtil.java | 6 +-
.../tajo/catalog/statistics/TableStats.java | 28 ++++----
.../src/main/proto/CatalogProtos.proto | 2 +-
.../org/apache/tajo/catalog/TestTableDesc.java | 2 +-
.../tajo/catalog/statistics/TestTableStat.java | 6 +-
.../tajo/master/DefaultTaskScheduler.java | 8 +--
.../tajo/master/querymaster/QueryUnit.java | 74 +++++---------------
.../master/querymaster/QueryUnitAttempt.java | 2 +-
.../tajo/master/querymaster/Repartitioner.java | 58 +++++++--------
.../tajo/master/querymaster/SubQuery.java | 16 ++---
.../java/org/apache/tajo/worker/Fetcher.java | 3 +-
.../main/java/org/apache/tajo/worker/Task.java | 2 +-
.../main/resources/webapps/worker/queryunit.jsp | 2 +-
.../org/apache/tajo/jdbc/TestResultSet.java | 3 +-
15 files changed, 85 insertions(+), 130 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e08f938..169be3f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -192,6 +192,9 @@ Release 0.8.0 - unreleased
BUG FIXES
+ TAJO-479: Rename obsolete name 'partition' to 'shuffle and fix the broken
+ taskdetail.jsp. (hyunsik).
+
TAJO-470: Fetcher's finished time and file length is changed in WEB UI.
(hyoungjunkim via hyunsik)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/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 fd72a9d..8e411b5 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
@@ -92,7 +92,7 @@ public class StatisticsUtil {
aggregated.setNumRows(aggregated.getNumRows() + ts.getNumRows());
aggregated.setNumBytes(aggregated.getNumBytes() + ts.getNumBytes());
aggregated.setNumBlocks(aggregated.getNumBlocks() + ts.getNumBlocks());
- aggregated.setNumPartitions(aggregated.getNumPartitions() + ts.getNumPartitions());
+ aggregated.setNumShuffleOutputs(aggregated.getNumShuffleOutputs() + ts.getNumShuffleOutputs());
}
//aggregated.setAvgRows(aggregated.getNumRows() / tableStats.size());
@@ -117,14 +117,14 @@ public class StatisticsUtil {
columnStatses.addAll(childStat.getColumnStats());
numBlocks += childStat.getNumBlocks();
numBytes += childStat.getNumBytes();
- numPartitions += childStat.getNumPartitions();
+ numPartitions += childStat.getNumShuffleOutputs();
numRows += childStat.getNumRows();
}
stat.setColumnStats(columnStatses);
stat.setNumBlocks(numBlocks);
stat.setNumBytes(numBytes);
- stat.setNumPartitions(numPartitions);
+ stat.setNumShuffleOutputs(numPartitions);
stat.setNumRows(numRows);
stat.setAvgRows(avgRows);
return stat;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/TableStats.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/TableStats.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/TableStats.java
index 0f7bc29..9a72da6 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/TableStats.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/TableStats.java
@@ -41,7 +41,7 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
@Expose private Long numRows = null; // required
@Expose private Long numBytes = null; // required
@Expose private Integer numBlocks = null; // optional
- @Expose private Integer numPartitions = null; // optional
+ @Expose private Integer numShuffleOutputs = null; // optional
@Expose private Long avgRows = null; // optional
@Expose private List<ColumnStats> columnStatses = null; // repeated
@@ -49,7 +49,7 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
numRows = 0l;
numBytes = 0l;
numBlocks = 0;
- numPartitions = 0;
+ numShuffleOutputs = 0;
avgRows = 0l;
columnStatses = TUtil.newList();
}
@@ -63,10 +63,10 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
} else {
this.numBlocks = 0;
}
- if (proto.hasNumPartitions()) {
- this.numPartitions = proto.getNumPartitions();
+ if (proto.hasNumShuffleOutputs()) {
+ this.numShuffleOutputs = proto.getNumShuffleOutputs();
} else {
- this.numPartitions = 0;
+ this.numShuffleOutputs = 0;
}
if (proto.hasAvgRows()) {
this.avgRows = proto.getAvgRows();
@@ -107,12 +107,12 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
this.numBlocks = numBlocks;
}
- public Integer getNumPartitions() {
- return this.numPartitions;
+ public Integer getNumShuffleOutputs() {
+ return this.numShuffleOutputs;
}
- public void setNumPartitions(int numPartitions) {
- this.numPartitions = numPartitions;
+ public void setNumShuffleOutputs(int numOutputs) {
+ this.numShuffleOutputs = numOutputs;
}
public Long getAvgRows() {
@@ -142,7 +142,7 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
boolean eq = this.numRows.equals(other.numRows);
eq = eq && this.numBytes.equals(other.numBytes);
eq = eq && TUtil.checkEquals(this.numBlocks, other.numBlocks);
- eq = eq && TUtil.checkEquals(this.numPartitions, other.numPartitions);
+ eq = eq && TUtil.checkEquals(this.numShuffleOutputs, other.numShuffleOutputs);
eq = eq && TUtil.checkEquals(this.avgRows, other.avgRows);
eq = eq && TUtil.checkEquals(this.columnStatses, other.columnStatses);
return eq;
@@ -153,7 +153,7 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
public int hashCode() {
return Objects.hashCode(numRows, numBytes,
- numBlocks, numPartitions, columnStatses);
+ numBlocks, numShuffleOutputs, columnStatses);
}
public Object clone() throws CloneNotSupportedException {
@@ -162,7 +162,7 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
stat.numRows = numRows != null ? numRows.longValue() : null;
stat.numBytes = numBytes != null ? numBytes.longValue() : null;
stat.numBlocks = numBlocks != null ? numBlocks.intValue() : null;
- stat.numPartitions = numPartitions != null ? numPartitions.intValue() : null;
+ stat.numShuffleOutputs = numShuffleOutputs != null ? numShuffleOutputs.intValue() : null;
stat.columnStatses = new ArrayList<ColumnStats>(this.columnStatses);
return stat;
@@ -192,8 +192,8 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
if (this.numBlocks != null) {
builder.setNumBlocks(this.numBlocks);
}
- if (this.numPartitions != null) {
- builder.setNumPartitions(this.numPartitions);
+ if (this.numShuffleOutputs != null) {
+ builder.setNumShuffleOutputs(this.numShuffleOutputs);
}
if (this.avgRows != null) {
builder.setAvgRows(this.avgRows);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/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 2d5119d..8180812 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -183,7 +183,7 @@ message TableStatsProto {
required int64 numRows = 1;
required int64 numBytes = 2;
optional int32 numBlocks = 4;
- optional int32 numPartitions = 5;
+ optional int32 numShuffleOutputs = 5;
optional int64 avgRows = 6;
repeated ColumnStatsProto colStat = 7;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestTableDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestTableDesc.java b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestTableDesc.java
index fec8e80..d6f75ba 100644
--- a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestTableDesc.java
+++ b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestTableDesc.java
@@ -52,7 +52,7 @@ public class TestTableDesc {
stats.setNumRows(957685);
stats.setNumBytes(1023234);
stats.setNumBlocks(3123);
- stats.setNumPartitions(5);
+ stats.setNumShuffleOutputs(5);
stats.setAvgRows(80000);
int numCols = 2;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/statistics/TestTableStat.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/statistics/TestTableStat.java b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/statistics/TestTableStat.java
index 3e8da9a..c258f16 100644
--- a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/statistics/TestTableStat.java
+++ b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/statistics/TestTableStat.java
@@ -33,7 +33,7 @@ public class TestTableStat {
stat.setNumRows(957685);
stat.setNumBytes(1023234);
stat.setNumBlocks(3123);
- stat.setNumPartitions(5);
+ stat.setNumShuffleOutputs(5);
stat.setAvgRows(80000);
int numCols = 3;
@@ -48,7 +48,7 @@ public class TestTableStat {
assertTrue(957685 == stat.getNumRows());
assertTrue(1023234 == stat.getNumBytes());
assertTrue(3123 == stat.getNumBlocks());
- assertTrue(5 == stat.getNumPartitions());
+ assertTrue(5 == stat.getNumShuffleOutputs());
assertTrue(80000 == stat.getAvgRows());
assertEquals(3, stat.getColumnStats().size());
for (int i = 0; i < numCols; i++) {
@@ -69,7 +69,7 @@ public class TestTableStat {
public void tableStatEquals(TableStats s1, TableStats s2) {
assertEquals(s1.getNumRows(), s2.getNumRows());
assertEquals(s1.getNumBlocks(), s2.getNumBlocks());
- assertEquals(s1.getNumPartitions(), s2.getNumPartitions());
+ assertEquals(s1.getNumShuffleOutputs(), s2.getNumShuffleOutputs());
assertEquals(s1.getAvgRows(), s2.getAvgRows());
assertEquals(s1.getColumnStats().size(), s2.getColumnStats().size());
for (int i = 0; i < s1.getColumnStats().size(); i++) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
index 860a466..5c986ec 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/DefaultTaskScheduler.java
@@ -178,9 +178,9 @@ public class DefaultTaskScheduler extends AbstractTaskScheduler {
if (context.isLeafQuery()) {
QueryUnitAttemptScheduleContext queryUnitContext = new QueryUnitAttemptScheduleContext();
QueryUnit task = SubQuery.newEmptyQueryUnit(context, queryUnitContext, subQuery, nextTaskId++);
- task.setFragment2(castEvent.getLeftFragment());
+ task.setFragment(castEvent.getLeftFragment());
if (castEvent.getRightFragment() != null) {
- task.setFragment2(castEvent.getRightFragment());
+ task.setFragment(castEvent.getRightFragment());
}
subQuery.getEventHandler().handle(new TaskEvent(task.getId(), TaskEventType.T_SCHEDULE));
} else {
@@ -195,9 +195,9 @@ public class DefaultTaskScheduler extends AbstractTaskScheduler {
QueryUnit task = SubQuery.newEmptyQueryUnit(context, queryUnitContext, subQuery, nextTaskId++);
for (Entry<String, List<URI>> eachFetch : fetches.entrySet()) {
task.addFetches(eachFetch.getKey(), eachFetch.getValue());
- task.setFragment2(fragmentsForNonLeafTask[0]);
+ task.setFragment(fragmentsForNonLeafTask[0]);
if (fragmentsForNonLeafTask[1] != null) {
- task.setFragment2(fragmentsForNonLeafTask[1]);
+ task.setFragment(fragmentsForNonLeafTask[1]);
}
}
subQuery.getEventHandler().handle(new TaskEvent(task.getId(), TaskEventType.T_SCHEDULE));
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
index 85dfa2e..b0b23bf 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
@@ -29,7 +29,6 @@ import org.apache.hadoop.yarn.state.*;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.QueryUnitId;
-import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.engine.planner.logical.*;
import org.apache.tajo.master.FragmentPair;
@@ -41,7 +40,6 @@ import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.util.TajoIdUtils;
import java.net.URI;
-import java.net.URISyntaxException;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.locks.Lock;
@@ -65,7 +63,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
private Map<String, Set<FragmentProto>> fragMap;
private Map<String, Set<URI>> fetchMap;
- private List<ShuffleFileOutput> partitions;
+ private List<ShuffleFileOutput> shuffleFileOutputs;
private TableStats stats;
private final boolean isLeafTask;
private List<IntermediateEntry> intermediateData;
@@ -127,7 +125,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
scan = new ArrayList<ScanNode>();
fetchMap = Maps.newHashMap();
fragMap = Maps.newHashMap();
- partitions = new ArrayList<ShuffleFileOutput>();
+ shuffleFileOutputs = new ArrayList<ShuffleFileOutput>();
attempts = Collections.emptyMap();
lastAttemptId = null;
nextAttempt = -1;
@@ -185,20 +183,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
}
}
- @Deprecated
- public void setFragment(String tableId, FileFragment fragment) {
- Set<FragmentProto> fragmentProtos;
- if (fragMap.containsKey(tableId)) {
- fragmentProtos = fragMap.get(tableId);
- } else {
- fragmentProtos = new HashSet<FragmentProto>();
- fragMap.put(tableId, fragmentProtos);
- }
- fragmentProtos.add(fragment.getProto());
- addDataLocation(fragment);
- }
-
- public void setFragment2(FileFragment fragment) {
+ public void setFragment(FileFragment fragment) {
Set<FragmentProto> fragmentProtos;
if (fragMap.containsKey(fragment.getTableName())) {
fragmentProtos = fragMap.get(fragment.getTableName());
@@ -212,9 +197,9 @@ public class QueryUnit implements EventHandler<TaskEvent> {
public void setFragment(FragmentPair[] fragmentPairs) {
for (FragmentPair eachFragmentPair : fragmentPairs) {
- this.setFragment2(eachFragmentPair.getLeftFragment());
+ this.setFragment(eachFragmentPair.getLeftFragment());
if (eachFragmentPair.getRightFragment() != null) {
- this.setFragment2(eachFragmentPair.getRightFragment());
+ this.setFragment(eachFragmentPair.getRightFragment());
}
}
}
@@ -226,21 +211,6 @@ public class QueryUnit implements EventHandler<TaskEvent> {
public String getSucceededHost() {
return succeededHost;
}
-
- public void addFetch(String tableId, String uri) throws URISyntaxException {
- this.addFetch(tableId, new URI(uri));
- }
-
- public void addFetch(String tableId, URI uri) {
- Set<URI> uris;
- if (fetchMap.containsKey(tableId)) {
- uris = fetchMap.get(tableId);
- } else {
- uris = Sets.newHashSet();
- }
- uris.add(uri);
- fetchMap.put(tableId, uris);
- }
public void addFetches(String tableId, Collection<URI> urilist) {
Set<URI> uris;
@@ -289,18 +259,6 @@ public class QueryUnit implements EventHandler<TaskEvent> {
public Collection<URI> getFetch(ScanNode scan) {
return this.fetchMap.get(scan.getTableName());
}
-
- public String getOutputName() {
- return this.store.getTableName();
- }
-
- public Schema getOutputSchema() {
- return this.store.getOutSchema();
- }
-
- public StoreTableNode getStoreTableNode() {
- return this.store;
- }
public ScanNode[] getScanNodes() {
return this.scan.toArray(new ScanNode[scan.size()]);
@@ -329,20 +287,20 @@ public class QueryUnit implements EventHandler<TaskEvent> {
this.stats = stats;
}
- public void setPartitions(List<ShuffleFileOutput> partitions) {
- this.partitions = Collections.unmodifiableList(partitions);
+ public void setShuffleFileOutputs(List<ShuffleFileOutput> partitions) {
+ this.shuffleFileOutputs = Collections.unmodifiableList(partitions);
}
public TableStats getStats() {
return this.stats;
}
- public List<ShuffleFileOutput> getPartitions() {
- return this.partitions;
+ public List<ShuffleFileOutput> getShuffleFileOutputs() {
+ return this.shuffleFileOutputs;
}
- public int getPartitionNum() {
- return this.partitions.size();
+ public int getShuffleOutpuNum() {
+ return this.shuffleFileOutputs.size();
}
public QueryUnitAttempt newAttempt() {
@@ -540,15 +498,15 @@ public class QueryUnit implements EventHandler<TaskEvent> {
public static class IntermediateEntry {
int taskId;
int attemptId;
- int partitionId;
+ int partId;
String pullHost;
int port;
- public IntermediateEntry(int taskId, int attemptId, int partitionId,
+ public IntermediateEntry(int taskId, int attemptId, int partId,
String pullServerAddr, int pullServerPort) {
this.taskId = taskId;
this.attemptId = attemptId;
- this.partitionId = partitionId;
+ this.partId = partId;
this.pullHost = pullServerAddr;
this.port = pullServerPort;
}
@@ -561,8 +519,8 @@ public class QueryUnit implements EventHandler<TaskEvent> {
return this.attemptId;
}
- public int getPartitionId() {
- return this.partitionId;
+ public int getPartId() {
+ return this.partId;
}
public String getPullHost() {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
index f65f810..f5001cc 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
@@ -188,7 +188,7 @@ public class QueryUnitAttempt implements EventHandler<TaskAttemptEvent> {
private void fillTaskStatistics(TaskCompletionReport report) {
if (report.getShuffleFileOutputsCount() > 0) {
- this.getQueryUnit().setPartitions(report.getShuffleFileOutputsList());
+ this.getQueryUnit().setShuffleFileOutputs(report.getShuffleFileOutputsList());
List<IntermediateEntry> partitions = new ArrayList<IntermediateEntry>();
for (ShuffleFileOutput p : report.getShuffleFileOutputsList()) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
index bec0620..2eed5d8 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
@@ -122,7 +122,7 @@ public class Repartitioner {
} else {
LOG.info("[Distributed Join Strategy] : Repartition Join");
// The hash map is modeling as follows:
- // <Partition Id, <Table Name, Intermediate Data>>
+ // <Part Id, <Table Name, Intermediate Data>>
Map<Integer, Map<String, List<IntermediateEntry>>> hashEntries = new HashMap<Integer, Map<String, List<IntermediateEntry>>>();
// Grouping IntermediateData by a partition key and a table name
@@ -131,9 +131,9 @@ public class Repartitioner {
for (QueryUnit task : childSubQuery.getQueryUnits()) {
if (task.getIntermediateData() != null) {
for (IntermediateEntry intermEntry : task.getIntermediateData()) {
- if (hashEntries.containsKey(intermEntry.getPartitionId())) {
+ if (hashEntries.containsKey(intermEntry.getPartId())) {
Map<String, List<IntermediateEntry>> tbNameToInterm =
- hashEntries.get(intermEntry.getPartitionId());
+ hashEntries.get(intermEntry.getPartId());
if (tbNameToInterm.containsKey(scan.getCanonicalName())) {
tbNameToInterm.get(scan.getCanonicalName()).add(intermEntry);
@@ -144,7 +144,7 @@ public class Repartitioner {
Map<String, List<IntermediateEntry>> tbNameToInterm =
new HashMap<String, List<IntermediateEntry>>();
tbNameToInterm.put(scan.getCanonicalName(), TUtil.newList(intermEntry));
- hashEntries.put(intermEntry.getPartitionId(), tbNameToInterm);
+ hashEntries.put(intermEntry.getPartId(), tbNameToInterm);
}
}
}
@@ -171,7 +171,7 @@ public class Repartitioner {
// determine the number of task per 64MB
int maxTaskNum = (int) Math.ceil((double)mb / desireJoinTaskVolumn);
LOG.info("The calculated number of tasks is " + maxTaskNum);
- LOG.info("The number of total partition keys is " + hashEntries.size());
+ LOG.info("The number of total shuffle keys is " + hashEntries.size());
// the number of join tasks cannot be larger than the number of
// distinct partition ids.
int joinTaskNum = Math.min(maxTaskNum, hashEntries.size());
@@ -182,7 +182,7 @@ public class Repartitioner {
// Assign partitions to tasks in a round robin manner.
for (Entry<Integer, Map<String, List<IntermediateEntry>>> entry
: hashEntries.entrySet()) {
- addJoinPartition(subQuery, entry.getKey(), entry.getValue());
+ addJoinShuffle(subQuery, entry.getKey(), entry.getValue());
}
schedulerContext.setTaskSize((int) Math.ceil((double) bothFetchSize / joinTaskNum));
@@ -209,13 +209,13 @@ public class Repartitioner {
schedulerContext.setEstimatedTaskNum(fragments.size());
}
- private static void addJoinPartition(SubQuery subQuery, int partitionId,
- Map<String, List<IntermediateEntry>> grouppedPartitions) {
+ private static void addJoinShuffle(SubQuery subQuery, int partitionId,
+ Map<String, List<IntermediateEntry>> grouppedPartitions) {
Map<String, List<URI>> fetches = new HashMap<String, List<URI>>();
for (ExecutionBlock execBlock : subQuery.getMasterPlan().getChilds(subQuery.getId())) {
Map<String, List<IntermediateEntry>> requests;
if (grouppedPartitions.containsKey(execBlock.getId().toString())) {
- requests = mergeHashPartitionRequest(grouppedPartitions.get(execBlock.getId().toString()));
+ requests = mergeHashShuffleRequest(grouppedPartitions.get(execBlock.getId().toString()));
} else {
return;
}
@@ -238,10 +238,8 @@ public class Repartitioner {
*
* @return key: pullserver's address, value: a list of requests
*/
- private static Map<String, List<IntermediateEntry>> mergeHashPartitionRequest(
- List<IntermediateEntry> partitions) {
- Map<String, List<IntermediateEntry>> mergedPartitions =
- new HashMap<String, List<IntermediateEntry>>();
+ private static Map<String, List<IntermediateEntry>> mergeHashShuffleRequest(List<IntermediateEntry> partitions) {
+ Map<String, List<IntermediateEntry>> mergedPartitions = new HashMap<String, List<IntermediateEntry>>();
for (IntermediateEntry partition : partitions) {
if (mergedPartitions.containsKey(partition.getPullAddress())) {
mergedPartitions.get(partition.getPullAddress()).add(partition);
@@ -258,16 +256,16 @@ public class Repartitioner {
DataChannel channel, int maxNum)
throws InternalException {
if (channel.getShuffleType() == HASH_SHUFFLE) {
- scheduleHashPartitionedFetches(schedulerContext, masterPlan, subQuery, channel, maxNum);
+ scheduleHashShuffledFetches(schedulerContext, masterPlan, subQuery, channel, maxNum);
} else if (channel.getShuffleType() == RANGE_SHUFFLE) {
- scheduleRangePartitionedFetches(schedulerContext, subQuery, childSubQuery, channel, maxNum);
+ scheduleRangeShuffledFetches(schedulerContext, subQuery, childSubQuery, channel, maxNum);
} else {
throw new InternalException("Cannot support partition type");
}
}
- public static void scheduleRangePartitionedFetches(TaskSchedulerContext schedulerContext, SubQuery subQuery,
- SubQuery childSubQuery, DataChannel channel, int maxNum)
+ public static void scheduleRangeShuffledFetches(TaskSchedulerContext schedulerContext, SubQuery subQuery,
+ SubQuery childSubQuery, DataChannel channel, int maxNum)
throws InternalException {
ExecutionBlock execBlock = subQuery.getBlock();
TableStats stat = childSubQuery.getTableStat();
@@ -341,13 +339,13 @@ public class Repartitioner {
LOG.error(e);
}
- schedulePartitionByRoundRobin(subQuery, map, scan.getTableName(), determinedTaskNum);
+ scheduleFetchesByRoundRobin(subQuery, map, scan.getTableName(), determinedTaskNum);
schedulerContext.setEstimatedTaskNum(determinedTaskNum);
}
- public static void schedulePartitionByRoundRobin(SubQuery subQuery, Map<?, Set<URI>> partitions,
- String tableName, int num) {
+ public static void scheduleFetchesByRoundRobin(SubQuery subQuery, Map<?, Set<URI>> partitions,
+ String tableName, int num) {
int i;
Map<String, List<URI>>[] fetchesArray = new Map[num];
for (i = 0; i < num; i++) {
@@ -379,9 +377,9 @@ public class Repartitioner {
return sb.toString();
}
- public static void scheduleHashPartitionedFetches(TaskSchedulerContext schedulerContext, MasterPlan masterPlan,
- SubQuery subQuery, DataChannel channel,
- int maxNum) {
+ public static void scheduleHashShuffledFetches(TaskSchedulerContext schedulerContext, MasterPlan masterPlan,
+ SubQuery subQuery, DataChannel channel,
+ int maxNum) {
ExecutionBlock execBlock = subQuery.getBlock();
List<TableStats> tableStatses = new ArrayList<TableStats>();
@@ -504,22 +502,20 @@ public class Repartitioner {
return fetchURLs;
}
- public static Map<Integer, List<IntermediateEntry>> hashByKey(
- List<IntermediateEntry> entries) {
+ public static Map<Integer, List<IntermediateEntry>> hashByKey(List<IntermediateEntry> entries) {
Map<Integer, List<IntermediateEntry>> hashed = new HashMap<Integer, List<IntermediateEntry>>();
for (IntermediateEntry entry : entries) {
- if (hashed.containsKey(entry.getPartitionId())) {
- hashed.get(entry.getPartitionId()).add(entry);
+ if (hashed.containsKey(entry.getPartId())) {
+ hashed.get(entry.getPartId()).add(entry);
} else {
- hashed.put(entry.getPartitionId(), TUtil.newList(entry));
+ hashed.put(entry.getPartId(), TUtil.newList(entry));
}
}
return hashed;
}
- public static Map<String, List<IntermediateEntry>> hashByHost(
- List<IntermediateEntry> entries) {
+ public static Map<String, List<IntermediateEntry>> hashByHost(List<IntermediateEntry> entries) {
Map<String, List<IntermediateEntry>> hashed = new HashMap<String, List<IntermediateEntry>>();
String hostName;
@@ -535,7 +531,7 @@ public class Repartitioner {
return hashed;
}
- public static SubQuery setPartitionNumberForTwoPhase(SubQuery subQuery, final int desiredNum, DataChannel channel) {
+ public static SubQuery setShuffleOutputNumForTwoPhase(SubQuery subQuery, final int desiredNum, DataChannel channel) {
ExecutionBlock execBlock = subQuery.getBlock();
Column[] keys = null;
// if the next query is join,
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
index ef3a11f..3e44959 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
@@ -388,7 +388,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
TableStats stat = new TableStats();
TableStats childStat;
long avgRows = 0, numBytes = 0, numRows = 0;
- int numBlocks = 0, numPartitions = 0;
+ int numBlocks = 0, numOutputs = 0;
List<ColumnStats> columnStatses = Lists.newArrayList();
MasterPlan masterPlan = subQuery.getMasterPlan();
@@ -401,14 +401,14 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
columnStatses.addAll(childStat.getColumnStats());
numBlocks += childStat.getNumBlocks();
numBytes += childStat.getNumBytes();
- numPartitions += childStat.getNumPartitions();
+ numOutputs += childStat.getNumShuffleOutputs();
numRows += childStat.getNumRows();
}
stat.setColumnStats(columnStatses);
stat.setNumBlocks(numBlocks);
stat.setNumBytes(numBytes);
- stat.setNumPartitions(numPartitions);
+ stat.setNumShuffleOutputs(numOutputs);
stat.setNumRows(numRows);
stat.setAvgRows(avgRows);
return stat;
@@ -511,7 +511,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
} else {
ExecutionBlock parent = subQuery.getMasterPlan().getParent(subQuery.getBlock());
DataChannel channel = subQuery.getMasterPlan().getChannel(subQuery.getId(), parent.getId());
- setRepartitionIfNecessary(subQuery, channel);
+ setShuffleIfNecessary(subQuery, channel);
initTaskScheduler(subQuery);
schedule(subQuery);
LOG.info(subQuery.getTaskScheduler().remainingScheduledObjectNum() + " objects are scheduled");
@@ -551,10 +551,10 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
* If a parent block requires a repartition operation, the method sets proper repartition
* methods and the number of partitions to a given subquery.
*/
- private static void setRepartitionIfNecessary(SubQuery subQuery, DataChannel channel) {
+ private static void setShuffleIfNecessary(SubQuery subQuery, DataChannel channel) {
if (channel.getShuffleType() != ShuffleType.NONE_SHUFFLE) {
- int numTasks = calculatePartitionNum(subQuery, channel);
- Repartitioner.setPartitionNumberForTwoPhase(subQuery, numTasks, channel);
+ int numTasks = calculateShuffleOutputNum(subQuery, channel);
+ Repartitioner.setShuffleOutputNumForTwoPhase(subQuery, numTasks, channel);
}
}
@@ -581,7 +581,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
* @param subQuery
* @return
*/
- public static int calculatePartitionNum(SubQuery subQuery, DataChannel channel) {
+ public static int calculateShuffleOutputNum(SubQuery subQuery, DataChannel channel) {
TajoConf conf = subQuery.context.getConf();
MasterPlan masterPlan = subQuery.getMasterPlan();
ExecutionBlock parent = masterPlan.getParent(subQuery.getBlock());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Fetcher.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Fetcher.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Fetcher.java
index 99286ce..360bebd 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Fetcher.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Fetcher.java
@@ -135,7 +135,7 @@ public class Fetcher {
// Shut down executor threads to exit.
bootstrap.releaseExternalResources();
-
+ finishTime = System.currentTimeMillis();
return file;
}
@@ -215,7 +215,6 @@ public class Fetcher {
LOG.info("Data fetch is done, but cannot get all data "
+ "(received/total: " + fileLength + "/" + length + ")");
}
- finishTime = System.currentTimeMillis();
} else {
fc.write(chunk.getContent().toByteBuffer());
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
index 820c883..1b72d13 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
@@ -185,7 +185,7 @@ public class Task {
LOG.info("==================================");
LOG.info("* Subquery " + request.getId() + " is initialized");
LOG.info("* InterQuery: " + interQuery
- + (interQuery ? ", Use " + this.shuffleType + " partitioning":""));
+ + (interQuery ? ", Use " + this.shuffleType + " shuffle":""));
LOG.info("* Fragments (num: " + request.getFragments().size() + ")");
LOG.info("* Fetches (total:" + request.getFetches().size() + ") :");
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/main/resources/webapps/worker/queryunit.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/resources/webapps/worker/queryunit.jsp b/tajo-core/tajo-core-backend/src/main/resources/webapps/worker/queryunit.jsp
index 61fa202..8317ee6 100644
--- a/tajo-core/tajo-core-backend/src/main/resources/webapps/worker/queryunit.jsp
+++ b/tajo-core/tajo-core-backend/src/main/resources/webapps/worker/queryunit.jsp
@@ -121,7 +121,7 @@
delim = "<br/>";
}
- int numPartitions = queryUnit.getPartitionNum();
+ int numPartitions = queryUnit.getShuffleOutpuNum();
String partitionKey = "-";
String partitionFileName = "-";
if(numPartitions > 0) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/f825dceb/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
index 0052827..b591a8c 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
@@ -28,7 +28,6 @@ import org.apache.tajo.TpchTestBase;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.catalog.statistics.TableStats;
-import org.apache.tajo.jdbc.TajoResultSet;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.DatumFactory;
@@ -87,7 +86,7 @@ public class TestResultSet {
stats.setNumBytes(written);
stats.setAvgRows(tupleNum);
stats.setNumBlocks(1000);
- stats.setNumPartitions(100);
+ stats.setNumShuffleOutputs(100);
desc = new TableDesc("score", scoreSchema, scoreMeta, p);
desc.setStats(stats);
}