You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2014/07/15 09:16:40 UTC
[01/19] git commit: TAJO-912: Tsql prints wrong version. (Mai Hai
Thanh via hyunsik)
Repository: tajo
Updated Branches:
refs/heads/index_support 31d524c51 -> e9207f3b3
TAJO-912: Tsql prints wrong version. (Mai Hai Thanh via hyunsik)
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/101413e3
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/101413e3
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/101413e3
Branch: refs/heads/index_support
Commit: 101413e319cd1be0e2570c6759e52d98aee89725
Parents: a5d14f8
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 13:04:54 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 13:04:54 2014 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../java/org/apache/tajo/cli/HelpCommand.java | 58 ++++++++++++++++----
.../java/org/apache/tajo/TajoConstants.java | 2 +-
3 files changed, 51 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/101413e3/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index e25928c..c0208aa 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,6 +82,8 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-912: Tsql prints wrong version. (Mai Hai Thanh via hyunsik)
+
TAJO-902: Unicode delimiter does not work correctly. (jinho)
TAJO-905: When to_date() parses some date without day, the result will be
http://git-wip-us.apache.org/repos/asf/tajo/blob/101413e3/tajo-client/src/main/java/org/apache/tajo/cli/HelpCommand.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/cli/HelpCommand.java b/tajo-client/src/main/java/org/apache/tajo/cli/HelpCommand.java
index 53bc404..8d908ec 100644
--- a/tajo-client/src/main/java/org/apache/tajo/cli/HelpCommand.java
+++ b/tajo-client/src/main/java/org/apache/tajo/cli/HelpCommand.java
@@ -18,11 +18,18 @@
package org.apache.tajo.cli;
-import org.apache.tajo.TajoConstants;
+import org.apache.tajo.util.VersionInfo;
+import java.io.IOException;
import java.io.PrintWriter;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.ProtocolException;
+import java.net.URL;
public class HelpCommand extends TajoShellCommand {
+ private String targetDocVersion = "";
+
public HelpCommand(TajoCli.TajoCliContext context) {
super(context);
}
@@ -34,7 +41,10 @@ public class HelpCommand extends TajoShellCommand {
@Override
public void invoke(String[] cmd) throws Exception {
- String docVersion = getDocumentationVersion();
+ if(targetDocVersion.equalsIgnoreCase("")) {
+ targetDocVersion = getDocumentationVersion();
+ }
+
PrintWriter sout = context.getOutput();
sout.println();
@@ -71,21 +81,49 @@ public class HelpCommand extends TajoShellCommand {
sout.println();
sout.println("Documentations");
- sout.println(" tsql guide http://tajo.apache.org/docs/"+ docVersion +"/cli.html");
- sout.println(" Query language http://tajo.apache.org/docs/"+ docVersion +"/sql_language.html");
- sout.println(" Functions http://tajo.apache.org/docs/"+ docVersion +"/functions.html");
- sout.println(" Backup & restore http://tajo.apache.org/docs/"+ docVersion +"/backup_and_restore.html");
- sout.println(" Configuration http://tajo.apache.org/docs/"+ docVersion +"/configuration.html");
+ sout.println(" tsql guide http://tajo.apache.org/docs/"+ targetDocVersion +"/cli.html");
+ sout.println(" Query language http://tajo.apache.org/docs/"+ targetDocVersion +"/sql_language.html");
+ sout.println(" Functions http://tajo.apache.org/docs/"+ targetDocVersion +"/functions.html");
+ sout.println(" Backup & restore http://tajo.apache.org/docs/"+ targetDocVersion +"/backup_and_restore.html");
+ sout.println(" Configuration http://tajo.apache.org/docs/"+ targetDocVersion +"/configuration.html");
sout.println();
}
private String getDocumentationVersion() {
- int delimiterIdx = TajoConstants.TAJO_VERSION.indexOf("-");
+ String tajoVersion = "", docVersion = "", docDefaultVersion = "current";
+ String tajoFullVersion = VersionInfo.getVersion();
+
+ int delimiterIdx = tajoFullVersion.indexOf("-");
if (delimiterIdx > -1) {
- return TajoConstants.TAJO_VERSION.substring(0, delimiterIdx);
+ tajoVersion = tajoFullVersion.substring(0, delimiterIdx);
} else {
- return TajoConstants.TAJO_VERSION;
+ tajoVersion = tajoFullVersion;
}
+
+ if(tajoVersion.equalsIgnoreCase("")) {
+ docVersion = docDefaultVersion;
+ } else {
+ try {
+ URL u = new URL("http://tajo.apache.org/docs/"+ tajoVersion + "/");
+ HttpURLConnection huc = (HttpURLConnection) u.openConnection();
+ huc.setConnectTimeout(1000);
+ huc.setReadTimeout(1000);
+ huc.setRequestMethod("HEAD");
+ if(huc.getResponseCode() == HttpURLConnection.HTTP_OK) {
+ docVersion = tajoVersion;
+ } else {
+ docVersion = docDefaultVersion;
+ }
+ } catch (MalformedURLException e0) {
+ docVersion = docDefaultVersion;
+ } catch (ProtocolException e1) {
+ docVersion = docDefaultVersion;
+ } catch (IOException e2) {
+ docVersion = docDefaultVersion;
+ }
+ }
+
+ return docVersion;
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/101413e3/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java b/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
index 028ca81..5f28f1c 100644
--- a/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
+++ b/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
@@ -19,7 +19,7 @@
package org.apache.tajo;
public class TajoConstants {
- public static final String TAJO_VERSION = "0.8.0-SNAPSHOT";
+ public static final String TAJO_VERSION = "0.9.0-SNAPSHOT";
public static final String SYSTEM_CONF_FILENAME = "system_conf.xml";
public static final String SYSTEM_DIR_NAME = "system";
[13/19] git commit: TAJO-916: SubQuery::computeStatFromTasks
occasionally fail. (Hyoungjun Kim via hyunsik)
Posted by ji...@apache.org.
TAJO-916: SubQuery::computeStatFromTasks occasionally fail. (Hyoungjun Kim via hyunsik)
Closes #69
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/fe557ad2
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/fe557ad2
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/fe557ad2
Branch: refs/heads/index_support
Commit: fe557ad24a70f6d698cc85f0bb75765906eb3641
Parents: 0a39818
Author: Hyunsik Choi <hy...@apache.org>
Authored: Mon Jul 14 18:19:07 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Mon Jul 14 18:19:30 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 ++
.../tajo/catalog/statistics/TableStats.java | 9 +---
.../tajo/catalog/statistics/TestTableStat.java | 55 ++++++++++++++++++++
3 files changed, 59 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/fe557ad2/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 26d3341..3c326dc 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,6 +82,9 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-916: SubQuery::computeStatFromTasks occasionally fail.
+ (Hyoungjun Kim via hyunsik)
+
TAJO-869: Sometimes, the unit test of testTaskRunnerHistory is failed.
(jinho)
http://git-wip-us.apache.org/repos/asf/tajo/blob/fe557ad2/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 c04545c..dd358ae 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
@@ -36,8 +36,6 @@ import java.util.ArrayList;
import java.util.List;
public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, GsonObject {
- private TableStatsProto.Builder builder = TableStatsProto.newBuilder();
-
@Expose private Long numRows = null; // required
@Expose private Long numBytes = null; // required
@Expose private Integer numBlocks = null; // optional
@@ -178,7 +176,6 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
public Object clone() throws CloneNotSupportedException {
TableStats stat = (TableStats) super.clone();
- stat.builder = CatalogProtos.TableStatsProto.newBuilder();
stat.numRows = numRows != null ? numRows : null;
stat.numBytes = numBytes != null ? numBytes : null;
stat.numBlocks = numBlocks != null ? numBlocks : null;
@@ -241,11 +238,7 @@ public class TableStats implements ProtoObject<TableStatsProto>, Cloneable, Gson
@Override
public TableStatsProto getProto() {
- if (builder == null) {
- builder = CatalogProtos.TableStatsProto.newBuilder();
- } else {
- builder.clear();
- }
+ TableStatsProto.Builder builder = CatalogProtos.TableStatsProto.newBuilder();
builder.setNumRows(this.numRows);
builder.setNumBytes(this.numBytes);
http://git-wip-us.apache.org/repos/asf/tajo/blob/fe557ad2/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 c258f16..b649dd7 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
@@ -20,11 +20,19 @@ package org.apache.tajo.catalog.statistics;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
+import org.apache.tajo.catalog.proto.CatalogProtos.TableStatsProto;
import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.datum.TextDatum;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
public class TestTableStat {
@Test
@@ -76,4 +84,51 @@ public class TestTableStat {
assertEquals(s1.getColumnStats().get(i), s2.getColumnStats().get(i));
}
}
+
+ @Test
+ public void testGetProtoThreadSafe() throws Exception {
+ final TableStats tableStats = new TableStats();
+
+ List<ColumnStats> columnStatsList = new ArrayList<ColumnStats>();
+ for (int i = 0; i < 3; i++) {
+ Column column = new Column("col_" + (i + 1), Type.TEXT);
+ ColumnStats columnStats = new ColumnStats(column);
+ columnStats.setMinValue(new TextDatum(i + ""));
+ columnStats.setMaxValue(new TextDatum((100 - i) + ""));
+
+ columnStatsList.add(columnStats);
+ }
+ tableStats.setColumnStats(columnStatsList);
+
+ int numThread = 10;
+ final CountDownLatch latch = new CountDownLatch(numThread);
+ final AtomicBoolean success = new AtomicBoolean(true);
+ for (int i = 0; i < numThread; i++) {
+ Thread t = new Thread() {
+ public void run() {
+ for (int j = 0; j < 100; j++) {
+ try {
+ TableStatsProto proto = tableStats.getProto();
+ if (tableStats.getColumnStats().size() != proto.getColStatList().size()) {
+ success.set(false);
+ break;
+ }
+ } catch (Exception e) {
+ success.set(false);
+ }
+ }
+
+ latch.countDown();
+ }
+ };
+
+ t.start();
+ }
+
+ latch.await();
+
+ if (!success.get()) {
+ fail("TableStats returns different column ststs");
+ }
+ }
}
[04/19] git commit: TAJO-925: Child ExecutionBlock of JOIN node has
different number of shuffle keys. (Hyoungjun Kim via hyunsik)
Posted by ji...@apache.org.
TAJO-925: Child ExecutionBlock of JOIN node has different number of shuffle keys. (Hyoungjun Kim via hyunsik)
Closes #61
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/438010f9
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/438010f9
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/438010f9
Branch: refs/heads/index_support
Commit: 438010f92bdbde50447d9fbc3438e57ddaff776f
Parents: 52c8600
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 14:42:18 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 14:42:18 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +
.../planner/global/ExecutionBlockCursor.java | 131 ++++++++++++++++++-
.../tajo/engine/planner/global/MasterPlan.java | 11 ++
.../apache/tajo/master/querymaster/Query.java | 15 ++-
.../tajo/master/querymaster/SubQuery.java | 22 ++--
.../apache/tajo/engine/query/TestJoinQuery.java | 48 +++++++
6 files changed, 212 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/438010f9/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 563d64e..e9e512d 100644
--- a/CHANGES
+++ b/CHANGES
@@ -90,6 +90,9 @@ Release 0.9.0 - unreleased
TAJO-912: Tsql prints wrong version. (Mai Hai Thanh via hyunsik)
+ TAJO-925: Child ExecutionBlock of JOIN node has different number of
+ shuffle keys. (Hyoungjun Kim via hyunsik)
+
TAJO-902: Unicode delimiter does not work correctly. (jinho)
TAJO-905: When to_date() parses some date without day, the result will be
http://git-wip-us.apache.org/repos/asf/tajo/blob/438010f9/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/ExecutionBlockCursor.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/ExecutionBlockCursor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/ExecutionBlockCursor.java
index d4ab068..0372769 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/ExecutionBlockCursor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/ExecutionBlockCursor.java
@@ -14,22 +14,35 @@
package org.apache.tajo.engine.planner.global;
-import java.util.ArrayList;
-import java.util.Stack;
+import org.apache.tajo.ExecutionBlockId;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* A distributed execution plan (DEP) is a direct acyclic graph (DAG) of ExecutionBlocks.
* This class is a pointer to an ExecutionBlock that the query engine should execute.
- * For each call of nextBlock(), it retrieves a next ExecutionBlock in a postfix order.
*/
public class ExecutionBlockCursor {
private MasterPlan masterPlan;
private ArrayList<ExecutionBlock> orderedBlocks = new ArrayList<ExecutionBlock>();
private int cursor = 0;
+ private List<BuildOrderItem> executionOrderedBlocks = new ArrayList<BuildOrderItem>();
+ private List<BuildOrderItem> notOrderedSiblingBlocks = new ArrayList<BuildOrderItem>();
+ private Map<ExecutionBlockId, AtomicInteger> orderRequiredChildCountMap = new HashMap<ExecutionBlockId, AtomicInteger>();
+
public ExecutionBlockCursor(MasterPlan plan) {
+ this(plan, false);
+ }
+
+ public ExecutionBlockCursor(MasterPlan plan, boolean siblingFirstOrder) {
this.masterPlan = plan;
- buildOrder(plan.getRoot());
+ if (siblingFirstOrder) {
+ buildSiblingFirstOrder(plan.getRoot());
+ } else {
+ buildDepthFirstOrder(plan.getRoot());
+ }
}
public int size() {
@@ -37,23 +50,127 @@ public class ExecutionBlockCursor {
}
// Add all execution blocks in a depth first and postfix order
- private void buildOrder(ExecutionBlock current) {
+ private void buildDepthFirstOrder(ExecutionBlock current) {
Stack<ExecutionBlock> stack = new Stack<ExecutionBlock>();
if (!masterPlan.isLeaf(current.getId())) {
for (ExecutionBlock execBlock : masterPlan.getChilds(current)) {
if (!masterPlan.isLeaf(execBlock)) {
- buildOrder(execBlock);
+ buildDepthFirstOrder(execBlock);
} else {
stack.push(execBlock);
}
}
for (ExecutionBlock execBlock : stack) {
- buildOrder(execBlock);
+ buildDepthFirstOrder(execBlock);
}
}
orderedBlocks.add(current);
}
+
+ private void buildSiblingFirstOrder(ExecutionBlock current) {
+ /*
+ |-eb_1404887024677_0004_000007
+ |-eb_1404887024677_0004_000006
+ |-eb_1404887024677_0004_000005
+ |-eb_1404887024677_0004_000004
+ |-eb_1404887024677_0004_000003
+ |-eb_1404887024677_0004_000002
+ |-eb_1404887024677_0004_000001
+
+ In the case of the upper plan, buildDepthFirstOrder() makes the following order in a depth first and postfix order.
+ [eb_1, eb_2, eb_3, eb_4, eb_5, eb_6, eb_7]
+ The eb_2 doesn't know eb_3's output bytes and uses a size of eb_4's all scan nodes.
+
+ buildSiblingFirstOrder() makes the following order in a sibling order.
+ [eb_1, eb_3, eb_2, eb_4, eb_5, eb_6, eb_7]
+ In this order the eb_2 knows eb_3's output bytes and the eb_4 also knows eb_1's output bytes.
+ */
+ preExecutionOrder(new BuildOrderItem(null, current));
+
+ for (BuildOrderItem eachItem: executionOrderedBlocks) {
+ if (masterPlan.isLeaf(eachItem.eb.getId())) {
+ orderedBlocks.add(eachItem.eb);
+ orderRequiredChildCountMap.get(eachItem.parentEB.getId()).decrementAndGet();
+ } else {
+ if (eachItem.allSiblingsOrdered()) {
+ for (BuildOrderItem eachSiblingItem: notOrderedSiblingBlocks) {
+ orderedBlocks.add(eachSiblingItem.eb);
+ }
+ orderedBlocks.add(eachItem.eb);
+ notOrderedSiblingBlocks.clear();
+ } else {
+ notOrderedSiblingBlocks.add(eachItem);
+ }
+ }
+ }
+ }
+
+ private void preExecutionOrder(BuildOrderItem current) {
+ Stack<BuildOrderItem> stack = new Stack<BuildOrderItem>();
+ if (!masterPlan.isLeaf(current.eb.getId())) {
+ List<ExecutionBlock> children = masterPlan.getChilds(current.eb);
+ orderRequiredChildCountMap.put(current.eb.getId(), new AtomicInteger(children.size()));
+ for (ExecutionBlock execBlock : children) {
+ BuildOrderItem item = new BuildOrderItem(current.eb, execBlock);
+ item.setSiblings(children);
+ if (!masterPlan.isLeaf(execBlock)) {
+ preExecutionOrder(item);
+ } else {
+ stack.push(item);
+ }
+ }
+ for (BuildOrderItem eachItem : stack) {
+ preExecutionOrder(eachItem);
+ }
+ }
+ executionOrderedBlocks.add(current);
+ }
+
+ class BuildOrderItem {
+ ExecutionBlock eb;
+ ExecutionBlock parentEB;
+ List<ExecutionBlockId> siblings = new ArrayList<ExecutionBlockId>();
+
+ BuildOrderItem(ExecutionBlock parentEB, ExecutionBlock eb) {
+ this.parentEB = parentEB;
+ this.eb = eb;
+ }
+
+ public void setSiblings(List<ExecutionBlock> siblings) {
+ for (ExecutionBlock eachEB: siblings) {
+ if (eachEB.getId().equals(eb.getId())) {
+ continue;
+ }
+
+ this.siblings.add(eachEB.getId());
+ }
+ }
+
+ public boolean allSiblingsOrdered() {
+ for (ExecutionBlockId eachSibling: siblings) {
+ if (orderRequiredChildCountMap.get(eachSibling) != null &&
+ orderRequiredChildCountMap.get(eachSibling).get() > 0) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return eb.toString();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof BuildOrderItem)) {
+ return false;
+ }
+ return eb.equals(((BuildOrderItem) obj).eb);
+ }
+ }
+
public boolean hasNext() {
return cursor < orderedBlocks.size();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/438010f9/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/MasterPlan.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/MasterPlan.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/MasterPlan.java
index a8593e5..9b99c5f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/MasterPlan.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/MasterPlan.java
@@ -212,6 +212,17 @@ public class MasterPlan {
sb.append(execBlockGraph.toStringGraph(getRoot().getId()));
sb.append("-------------------------------------------------------------------------------\n");
+ ExecutionBlockCursor executionOrderCursor = new ExecutionBlockCursor(this, true);
+ sb.append("Order of Execution\n");
+ sb.append("-------------------------------------------------------------------------------");
+ int order = 1;
+ while (executionOrderCursor.hasNext()) {
+ ExecutionBlock currentEB = executionOrderCursor.nextBlock();
+ sb.append("\n").append(order).append(": ").append(currentEB.getId());
+ order++;
+ }
+ sb.append("\n-------------------------------------------------------------------------------\n");
+
while(cursor.hasNext()) {
ExecutionBlock block = cursor.nextBlock();
http://git-wip-us.apache.org/repos/asf/tajo/blob/438010f9/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
index 0ce6d7e..31199ba 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
@@ -209,7 +209,19 @@ public class Query implements EventHandler<QueryEvent> {
this.eventHandler = eventHandler;
this.plan = plan;
this.sm = context.getStorageManager();
- cursor = new ExecutionBlockCursor(plan);
+ this.cursor = new ExecutionBlockCursor(plan, true);
+
+ StringBuilder sb = new StringBuilder("\n=======================================================");
+ sb.append("\nThe order of execution: \n");
+ int order = 1;
+ while (cursor.hasNext()) {
+ ExecutionBlock currentEB = cursor.nextBlock();
+ sb.append("\n").append(order).append(": ").append(currentEB.getId());
+ order++;
+ }
+ sb.append("\n=======================================================");
+ LOG.info(sb);
+ cursor.reset();
ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
this.readLock = readWriteLock.readLock();
@@ -340,6 +352,7 @@ public class Query implements EventHandler<QueryEvent> {
@Override
public void transition(Query query, QueryEvent queryEvent) {
+
query.setStartTime();
SubQuery subQuery = new SubQuery(query.context, query.getPlan(),
query.getExecutionBlockCursor().nextBlock(), query.sm);
http://git-wip-us.apache.org/repos/asf/tajo/blob/438010f9/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
index 0776722..94f8b32 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
@@ -746,35 +746,37 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
int totalMem = getClusterTotalMemory(subQuery);
LOG.info(subQuery.getId() + ", Total memory of cluster is " + totalMem + " MB");
int slots = Math.max(totalMem / conf.getIntVar(ConfVars.TASK_DEFAULT_MEMORY), 1);
-
// determine the number of task
taskNum = Math.min(taskNum, slots);
+
if (conf.getIntVar(ConfVars.TESTCASE_MIN_TASK_NUM) > 0) {
taskNum = conf.getIntVar(ConfVars.TESTCASE_MIN_TASK_NUM);
LOG.warn("!!!!! TESTCASE MODE !!!!!");
}
- LOG.info(subQuery.getId() + ", The determined number of join partitions is " + taskNum);
// The shuffle output numbers of join may be inconsistent by execution block order.
// Thus, we need to compare the number with DataChannel output numbers.
// If the number is right, the number and DataChannel output numbers will be consistent.
- int outerShuffleOutptNum = 0, innerShuffleOutputNum = 0;
+ int outerShuffleOutputNum = 0, innerShuffleOutputNum = 0;
for (DataChannel eachChannel : masterPlan.getOutgoingChannels(outer.getId())) {
- outerShuffleOutptNum = Math.max(outerShuffleOutptNum, eachChannel.getShuffleOutputNum());
+ outerShuffleOutputNum = Math.max(outerShuffleOutputNum, eachChannel.getShuffleOutputNum());
}
-
for (DataChannel eachChannel : masterPlan.getOutgoingChannels(inner.getId())) {
innerShuffleOutputNum = Math.max(innerShuffleOutputNum, eachChannel.getShuffleOutputNum());
}
-
- if (outerShuffleOutptNum != innerShuffleOutputNum
- && taskNum != outerShuffleOutptNum
+ if (outerShuffleOutputNum != innerShuffleOutputNum
+ && taskNum != outerShuffleOutputNum
&& taskNum != innerShuffleOutputNum) {
- taskNum = Math.max(outerShuffleOutptNum, innerShuffleOutputNum);
+ LOG.info(subQuery.getId() + ", Change determined number of join partitions cause difference of outputNum" +
+ ", originTaskNum=" + taskNum + ", changedTaskNum=" + Math.max(outerShuffleOutputNum, innerShuffleOutputNum) +
+ ", outerShuffleOutptNum=" + outerShuffleOutputNum +
+ ", innerShuffleOutputNum=" + innerShuffleOutputNum);
+ taskNum = Math.max(outerShuffleOutputNum, innerShuffleOutputNum);
}
- return taskNum;
+ LOG.info(subQuery.getId() + ", The determined number of join partitions is " + taskNum);
+ return taskNum;
// Is this subquery the first step of group-by?
} else if (grpNode != null) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/438010f9/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
index 13a0b2b..ca1ece1 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
@@ -35,8 +35,10 @@ import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import java.sql.ResultSet;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.List;
import static org.junit.Assert.assertEquals;
@@ -1051,4 +1053,50 @@ public class TestJoinQuery extends QueryTestCaseBase {
cleanupQuery(res);
}
}
+
+ @Test
+ public void testJoinWithDifferentShuffleKey() throws Exception {
+ KeyValueSet tableOptions = new KeyValueSet();
+ tableOptions.put(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.put(StorageConstants.CSVFILE_NULL, "\\\\N");
+
+ Schema schema = new Schema();
+ schema.addColumn("id", Type.INT4);
+ schema.addColumn("name", Type.TEXT);
+
+ List<String> data = new ArrayList<String>();
+
+ int bytes = 0;
+ for (int i = 0; i < 1000000; i++) {
+ String row = i + "|" + i + "name012345678901234567890123456789012345678901234567890";
+ bytes += row.getBytes().length;
+ data.add(row);
+ if (bytes > 2 * 1024 * 1024) {
+ break;
+ }
+ }
+ TajoTestingCluster.createTable("large_table", schema, tableOptions, data.toArray(new String[]{}));
+
+ int originConfValue = conf.getIntVar(ConfVars.DIST_QUERY_JOIN_PARTITION_VOLUME);
+ testingCluster.setAllTajoDaemonConfValue(ConfVars.DIST_QUERY_JOIN_PARTITION_VOLUME.varname, "1");
+ ResultSet res = executeString(
+ "select count(b.id) " +
+ "from (select id, count(*) as cnt from large_table group by id) a " +
+ "left outer join (select id, count(*) as cnt from large_table where id < 200 group by id) b " +
+ "on a.id = b.id"
+ );
+
+ try {
+ String expected =
+ "?count\n" +
+ "-------------------------------\n" +
+ "200\n";
+
+ assertEquals(expected, resultSetToString(res));
+ } finally {
+ testingCluster.setAllTajoDaemonConfValue(ConfVars.DIST_QUERY_JOIN_PARTITION_VOLUME.varname, "" + originConfValue);
+ cleanupQuery(res);
+ executeString("DROP TABLE large_table PURGE").close();
+ }
+ }
}
[06/19] git commit: TAJO-929: Broadcast join with empty outer join
table returns empty result. (Hyoungjun Kim via hyunsik)
Posted by ji...@apache.org.
TAJO-929: Broadcast join with empty outer join table returns empty result. (Hyoungjun Kim via hyunsik)
Closes #66
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/329e508c
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/329e508c
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/329e508c
Branch: refs/heads/index_support
Commit: 329e508caf88b3f75e465581599a255eceabb16a
Parents: 42f3b4d
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 15:13:05 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 15:13:11 2014 +0900
----------------------------------------------------------------------
CHANGES | 5 +++-
.../tajo/master/querymaster/Repartitioner.java | 11 ++++++---
.../tajo/engine/query/TestJoinBroadcast.java | 26 ++++++++++++++++++++
3 files changed, 38 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/329e508c/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 0694eba..921183e 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,7 +82,10 @@ Release 0.9.0 - unreleased
BUG FIXES
- TAJO-927: Broadcast Join with Large, Small, Large, Small tables
+ TAJO-929: Broadcast join with empty outer join table returns empty result.
+ (Hyoungjun Kim via hyunsik)
+
+ TAJO-927: Broadcast Join with Large, Small, Large, Small tables
makes a wrong plan. (Hyoungjun Kim via hyunsik)
TAJO-913: Add some missed tests for constant value group-by keys.
http://git-wip-us.apache.org/repos/asf/tajo/blob/329e508c/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
index fc80bd1..ce2194e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
@@ -119,13 +119,18 @@ public class Repartitioner {
}
}
- // If one of inner join tables has no input data, it should return zero rows.
+ // If one of inner join tables has no input data, it means that this execution block has no result row.
JoinNode joinNode = PlannerUtil.findMostBottomNode(execBlock.getPlan(), NodeType.JOIN);
if (joinNode != null) {
if ( (joinNode.getJoinType() == JoinType.INNER)) {
+ LogicalNode leftNode = joinNode.getLeftChild();
+ LogicalNode rightNode = joinNode.getRightChild();
for (int i = 0; i < stats.length; i++) {
- if (stats[i] == 0) {
- return;
+ if (scans[i].getPID() == leftNode.getPID() || scans[i].getPID() == rightNode.getPID()) {
+ if (stats[i] == 0) {
+ LOG.info(scans[i] + " 's input data is zero. Inner join's result is empty.");
+ return;
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/329e508c/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
index 668992b..e01b3c5 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
@@ -504,6 +504,32 @@ public class TestJoinBroadcast extends QueryTestCaseBase {
}
}
+ @Test
+ public final void testInnerAndOuterWithEmpty() throws Exception {
+ executeDDL("customer_partition_ddl.sql", null);
+ executeFile("insert_into_customer_partition.sql").close();
+
+ // outer join table is empty
+ ResultSet res = executeString(
+ "select a.l_orderkey, b.o_orderkey, c.c_custkey from lineitem a " +
+ "inner join orders b on a.l_orderkey = b.o_orderkey " +
+ "left outer join customer_broad_parts c on a.l_orderkey = c.c_custkey and c.c_custkey < 0"
+ );
+
+ String expected = "l_orderkey,o_orderkey,c_custkey\n" +
+ "-------------------------------\n" +
+ "1,1,null\n" +
+ "1,1,null\n" +
+ "2,2,null\n" +
+ "3,3,null\n" +
+ "3,3,null\n";
+
+ assertEquals(expected, resultSetToString(res));
+ res.close();
+
+ executeString("DROP TABLE customer_broad_parts PURGE").close();
+ }
+
static interface TupleCreator {
public Tuple createTuple(String[] columnDatas);
}
[16/19] git commit: TAJO-936: TestStorages::testSplitable is failed
occasionally. (jinho)
Posted by ji...@apache.org.
TAJO-936: TestStorages::testSplitable is failed occasionally. (jinho)
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/aee7874f
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/aee7874f
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/aee7874f
Branch: refs/heads/index_support
Commit: aee7874faa45e3ba2eda3157683cf20111cacea5
Parents: 48dbfd9
Author: jinossy <ji...@gmail.com>
Authored: Tue Jul 15 14:25:51 2014 +0900
Committer: jinossy <ji...@gmail.com>
Committed: Tue Jul 15 14:25:51 2014 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../org/apache/tajo/storage/rcfile/RCFile.java | 1 +
.../org/apache/tajo/storage/TestStorages.java | 58 +++++++++++++++++++-
3 files changed, 60 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/aee7874f/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 3803ded..473db19 100644
--- a/CHANGES
+++ b/CHANGES
@@ -84,6 +84,8 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-936: TestStorages::testSplitable is failed occasionally. (jinho)
+
TAJO-673: Assign proper number of tasks when inserting into partitioned table. (jaehwa)
TAJO-916: SubQuery::computeStatFromTasks occasionally fail.
http://git-wip-us.apache.org/repos/asf/tajo/blob/aee7874f/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java b/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
index 1da41e4..4cc37b3 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
@@ -1380,6 +1380,7 @@ public class RCFile {
in.readFully(sync); // read sync bytes
headerEnd = in.getPos();
+ lastSeenSyncPos = headerEnd; //initial sync position
readBytes += headerEnd;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/aee7874f/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java b/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
index 944aa5c..1cf1ecf 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
@@ -23,7 +23,9 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoIdProtos;
-import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.common.TajoDataTypes.Type;
@@ -175,6 +177,60 @@ public class TestStorages {
}
@Test
+ public void testRCFileSplitable() throws IOException {
+ if (storeType == StoreType.RCFILE) {
+ Schema schema = new Schema();
+ schema.addColumn("id", Type.INT4);
+ schema.addColumn("age", Type.INT8);
+
+ TableMeta meta = CatalogUtil.newTableMeta(storeType);
+ Path tablePath = new Path(testDir, "Splitable.data");
+ Appender appender = StorageManagerFactory.getStorageManager(conf).getAppender(meta, schema, tablePath);
+ appender.enableStats();
+ appender.init();
+ int tupleNum = 10000;
+ VTuple vTuple;
+
+ for (int i = 0; i < tupleNum; i++) {
+ vTuple = new VTuple(2);
+ vTuple.put(0, DatumFactory.createInt4(i + 1));
+ vTuple.put(1, DatumFactory.createInt8(25l));
+ appender.addTuple(vTuple);
+ }
+ appender.close();
+ TableStats stat = appender.getStats();
+ assertEquals(tupleNum, stat.getNumRows().longValue());
+
+ FileStatus status = fs.getFileStatus(tablePath);
+ long fileLen = status.getLen();
+ long randomNum = 122; // header size
+
+ FileFragment[] tablets = new FileFragment[2];
+ tablets[0] = new FileFragment("Splitable", tablePath, 0, randomNum);
+ tablets[1] = new FileFragment("Splitable", tablePath, randomNum, (fileLen - randomNum));
+
+ Scanner scanner = StorageManagerFactory.getStorageManager(conf).getScanner(meta, schema, tablets[0], schema);
+ assertTrue(scanner.isSplittable());
+ scanner.init();
+ int tupleCnt = 0;
+ while (scanner.next() != null) {
+ tupleCnt++;
+ }
+ scanner.close();
+
+ scanner = StorageManagerFactory.getStorageManager(conf).getScanner(meta, schema, tablets[1], schema);
+ assertTrue(scanner.isSplittable());
+ scanner.init();
+ while (scanner.next() != null) {
+ tupleCnt++;
+ }
+ scanner.close();
+
+ assertEquals(tupleNum, tupleCnt);
+ }
+ }
+
+ @Test
public void testProjection() throws IOException {
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
[19/19] git commit: Merge branch 'master' of
http://git-wip-us.apache.org/repos/asf/tajo into index_support
Posted by ji...@apache.org.
Merge branch 'master' of http://git-wip-us.apache.org/repos/asf/tajo into index_support
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/e9207f3b
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/e9207f3b
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/e9207f3b
Branch: refs/heads/index_support
Commit: e9207f3b3f63edf57731242e26edc443abd2e062
Parents: 31d524c 9cf1071
Author: Jihoon Son <ji...@apache.org>
Authored: Tue Jul 15 16:16:13 2014 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Tue Jul 15 16:16:13 2014 +0900
----------------------------------------------------------------------
CHANGES | 46 ++
.../tajo/catalog/statistics/ColumnStats.java | 10 +-
.../tajo/catalog/statistics/TableStats.java | 9 +-
.../tajo/catalog/statistics/TestTableStat.java | 55 ++
.../tajo-catalog-drivers/tajo-hcatalog/pom.xml | 4 +-
.../tajo/catalog/store/HCatalogStore.java | 25 +-
.../catalog/store/HCatalogStoreClientPool.java | 32 +-
.../java/org/apache/tajo/cli/HelpCommand.java | 58 ++-
.../java/org/apache/tajo/TajoConstants.java | 2 +-
.../java/org/apache/tajo/conf/TajoConf.java | 6 +
.../java/org/apache/tajo/util/KeyValueSet.java | 2 +-
tajo-common/src/main/proto/tajo_protos.proto | 1 +
.../tajo/engine/planner/LogicalPlanner.java | 3 +-
.../engine/planner/PhysicalPlannerImpl.java | 1 +
.../engine/planner/RangePartitionAlgorithm.java | 2 +-
.../planner/global/ExecutionBlockCursor.java | 131 ++++-
.../engine/planner/global/GlobalPlanner.java | 16 +-
.../tajo/engine/planner/global/MasterPlan.java | 11 +
.../planner/physical/ColPartitionStoreExec.java | 9 +-
.../DistinctGroupbyHashAggregationExec.java | 30 +-
.../planner/physical/HashLeftOuterJoinExec.java | 41 +-
.../physical/HashShuffleFileWriteExec.java | 1 +
.../planner/rewrite/ProjectionPushDownRule.java | 4 +-
.../org/apache/tajo/engine/utils/TupleUtil.java | 70 ++-
.../apache/tajo/master/querymaster/Query.java | 15 +-
.../tajo/master/querymaster/QueryUnit.java | 13 +
.../master/querymaster/QueryUnitAttempt.java | 2 +-
.../tajo/master/querymaster/Repartitioner.java | 143 +++++-
.../tajo/master/querymaster/SubQuery.java | 22 +-
.../java/org/apache/tajo/worker/Fetcher.java | 119 +++--
.../main/java/org/apache/tajo/worker/Task.java | 47 +-
.../apache/tajo/worker/TaskAttemptContext.java | 18 +
.../apache/tajo/worker/TaskRunnerHistory.java | 5 +-
.../src/main/proto/TajoWorkerProtocol.proto | 2 +
.../java/org/apache/tajo/QueryTestCaseBase.java | 35 +-
.../planner/global/TestBroadcastJoinPlan.java | 84 ++++
.../tajo/engine/query/TestCreateTable.java | 199 +++++---
.../tajo/engine/query/TestGroupByQuery.java | 85 +++-
.../tajo/engine/query/TestJoinBroadcast.java | 26 +
.../apache/tajo/engine/query/TestJoinQuery.java | 56 +++
.../apache/tajo/engine/query/TestSortQuery.java | 49 ++
.../tajo/engine/query/TestTablePartitions.java | 98 +++-
.../org/apache/tajo/jdbc/TestResultSet.java | 3 +
.../querymaster/TestQueryUnitStatusUpdate.java | 4 +-
.../org/apache/tajo/worker/TestFetcher.java | 162 ++++--
.../testGroupByWithConstantKeys2.sql | 1 +
.../testJoinFilterOfRowPreservedTable1.sql | 8 +
.../testGroupByWithConstantKeys1.result | 3 +
.../testGroupByWithConstantKeys2.result | 5 +
.../testJoinFilterOfRowPreservedTable1.result | 15 +
.../java/org/apache/tajo/rpc/TestAsyncRpc.java | 22 +-
tajo-storage/pom.xml | 4 +-
.../org/apache/tajo/storage/RowStoreUtil.java | 1 +
.../apache/tajo/storage/TableStatistics.java | 5 +-
.../tajo/storage/parquet/TajoParquetReader.java | 2 +-
.../tajo/storage/parquet/TajoParquetWriter.java | 2 +-
.../org/apache/tajo/storage/rcfile/RCFile.java | 1 +
.../thirdparty/parquet/CodecFactory.java | 196 ++++++++
.../parquet/ColumnChunkPageWriteStore.java | 211 ++++++++
.../parquet/InternalParquetRecordReader.java | 187 +++++++
.../parquet/InternalParquetRecordWriter.java | 156 ++++++
.../thirdparty/parquet/ParquetFileWriter.java | 504 +++++++++++++++++++
.../thirdparty/parquet/ParquetReader.java | 151 ++++++
.../thirdparty/parquet/ParquetWriter.java | 220 ++++++++
.../java/org/apache/tajo/storage/v2/RCFile.java | 5 -
.../org/apache/tajo/storage/TestStorages.java | 58 ++-
.../tajo/pullserver/PullServerAuxService.java | 4 +-
.../tajo/pullserver/TajoPullServerService.java | 32 +-
68 files changed, 3220 insertions(+), 329 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e9207f3b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e9207f3b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e9207f3b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e9207f3b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e9207f3b/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e9207f3b/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e9207f3b/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
----------------------------------------------------------------------
[03/19] git commit: TAJO-913: Add some missed tests for constant
value group-by keys.
Posted by ji...@apache.org.
TAJO-913: Add some missed tests for constant value group-by keys.
Closes #60
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/52c8600d
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/52c8600d
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/52c8600d
Branch: refs/heads/index_support
Commit: 52c8600dd59247b9be95f338ade4ea5aee2fba06
Parents: 29db645
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 14:30:38 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 14:30:38 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +++
.../apache/tajo/engine/query/TestGroupByQuery.java | 17 +++++++++++++++++
.../testGroupByWithConstantKeys2.sql | 1 +
.../testGroupByWithConstantKeys1.result | 3 +++
.../testGroupByWithConstantKeys2.result | 5 +++++
5 files changed, 29 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/52c8600d/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index f73bf1b..563d64e 100644
--- a/CHANGES
+++ b/CHANGES
@@ -81,6 +81,9 @@ Release 0.9.0 - unreleased
(Hyoungjun Kim via hyunsik)
BUG FIXES
+
+ TAJO-913: Add some missed tests for constant value group-by keys.
+ (hyunsik)
TAJO-909: {SortBased, Col}PartitionStoreExec should not write partition
keys to files. (Hyoungjun Kim via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/52c8600d/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
index 9f06317..40ee54e 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
@@ -139,6 +139,23 @@ public class TestGroupByQuery extends QueryTestCaseBase {
}
@Test
+ public final void testGroupByWithConstantKeys1() throws Exception {
+ // select 123 as key, count(1) as total from lineitem group by key order by key, total;
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
+
+ @Test
+ public final void testGroupByWithConstantKeys2() throws Exception {
+ // select l_partkey as a, timestamp '2014-07-07 04:28:31.561' as b, '##' as c, count(*) d from lineitem
+ // group by a, b, c;
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
+
+ @Test
public final void testDistinctAggregation1() throws Exception {
// select l_orderkey, max(l_orderkey) as maximum, count(distinct l_linenumber) as unique_key from lineitem
// group by l_orderkey;
http://git-wip-us.apache.org/repos/asf/tajo/blob/52c8600d/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql
new file mode 100644
index 0000000..439d049
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql
@@ -0,0 +1 @@
+select l_partkey as a, timestamp '2014-07-07 04:28:31.561' as b, '##' as c, count(*) d from lineitem group by a, b, c;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/52c8600d/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys1.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys1.result b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys1.result
new file mode 100644
index 0000000..2c04315
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys1.result
@@ -0,0 +1,3 @@
+key,total
+-------------------------------
+123,5
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/52c8600d/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result
new file mode 100644
index 0000000..e2a3b89
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result
@@ -0,0 +1,5 @@
+a,b,c,d
+-------------------------------
+3,2014-07-07 04:28:31.561,##,1
+1,2014-07-07 04:28:31.561,##,2
+2,2014-07-07 04:28:31.561,##,2
\ No newline at end of file
[07/19] git commit: TAJO-934: Multiple DISTINCT returns null grouping
key value. (Hyoungjun Kim via hyunsik)
Posted by ji...@apache.org.
TAJO-934: Multiple DISTINCT returns null grouping key value. (Hyoungjun Kim via hyunsik)
Closes #63
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/95292d29
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/95292d29
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/95292d29
Branch: refs/heads/index_support
Commit: 95292d29d96963c769ce2fd17a3350375145683e
Parents: 329e508
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 15:52:59 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 15:52:59 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +
.../tajo/engine/planner/LogicalPlanner.java | 3 +-
.../DistinctGroupbyHashAggregationExec.java | 30 ++++++---
.../tajo/engine/query/TestGroupByQuery.java | 68 +++++++++++++++++++-
.../java/org/apache/tajo/storage/v2/RCFile.java | 5 --
5 files changed, 93 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/95292d29/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 921183e..43075f0 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,6 +82,9 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-934: Multiple DISTINCT returns null grouping key value.
+ (Hyoungjun Kim via hyunsik)
+
TAJO-929: Broadcast join with empty outer join table returns empty result.
(Hyoungjun Kim via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/95292d29/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
index 80390d3..ea517c0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
@@ -676,7 +676,8 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
for (Iterator<NamedExpr> it = block.namedExprsMgr.getIteratorForUnevaluatedExprs(); it.hasNext();) {
NamedExpr rawTarget = it.next();
try {
- includeDistinctFunction = PlannerUtil.existsDistinctAggregationFunction(rawTarget.getExpr());
+ // check if at least distinct aggregation function
+ includeDistinctFunction |= PlannerUtil.existsDistinctAggregationFunction(rawTarget.getExpr());
EvalNode evalNode = exprAnnotator.createEvalNode(context.plan, context.queryBlock, rawTarget.getExpr());
if (evalNode.getType() == EvalType.AGG_FUNCTION) {
aggEvalNames.add(rawTarget.getAlias());
http://git-wip-us.apache.org/repos/asf/tajo/blob/95292d29/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
index 1a4b706..3fac509 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
@@ -114,6 +114,7 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
}
if (first) {
loadChildHashTable();
+
progress = 0.5f;
first = false;
}
@@ -141,9 +142,12 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
//--------------------------------------------------------------------------------------
List<List<Tuple>> tupleSlots = new ArrayList<List<Tuple>>();
+
+ // aggregation with single grouping key
for (int i = 0; i < hashAggregators.length; i++) {
if (!hashAggregators[i].iterator.hasNext()) {
nullCount++;
+ tupleSlots.add(new ArrayList<Tuple>());
continue;
}
Entry<Tuple, Map<Tuple, FunctionContext[]>> entry = hashAggregators[i].iterator.next();
@@ -158,10 +162,10 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
finished = true;
progress = 1.0f;
- // If DistinctGroupbyHashAggregationExec didn't has any rows,
+ // If DistinctGroupbyHashAggregationExec does not have any rows,
// it should return NullDatum.
if (totalNumRows == 0 && groupbyNodeNum == 0) {
- Tuple tuple = new VTuple(hashAggregators.length);
+ Tuple tuple = new VTuple(outputColumnNum);
for (int i = 0; i < tuple.size(); i++) {
tuple.put(i, DatumFactory.createNullDatum());
}
@@ -199,9 +203,11 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
*/
+ // currentAggregatedTuples has tuples which has same group key.
currentAggregatedTuples = new ArrayList<Tuple>();
int listIndex = 0;
while (true) {
+ // Each item in tuples is VTuple. So the tuples variable is two dimensions(tuple[aggregator][datum]).
Tuple[] tuples = new Tuple[hashAggregators.length];
for (int i = 0; i < hashAggregators.length; i++) {
List<Tuple> aggregatedTuples = tupleSlots.get(i);
@@ -212,7 +218,7 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
//merge
Tuple mergedTuple = new VTuple(outputColumnNum);
- int mergeTupleIndex = 0;
+ int resultColumnIdx = 0;
boolean allNull = true;
for (int i = 0; i < hashAggregators.length; i++) {
@@ -222,14 +228,22 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
int tupleSize = hashAggregators[i].getTupleSize();
for (int j = 0; j < tupleSize; j++) {
- if (resultColumnIdIndexes[mergeTupleIndex] >= 0) {
- if (tuples[i] != null) {
- mergedTuple.put(resultColumnIdIndexes[mergeTupleIndex], tuples[i].get(j));
+ int mergeTupleIndex = resultColumnIdIndexes[resultColumnIdx];
+ if (mergeTupleIndex >= 0) {
+ if (mergeTupleIndex < distinctGroupingKey.size()) {
+ // set group key tuple
+ // Because each hashAggregator has different number of tuples,
+ // sometimes getting group key from each hashAggregator will be null value.
+ mergedTuple.put(mergeTupleIndex, distinctGroupingKey.get(mergeTupleIndex));
} else {
- mergedTuple.put(resultColumnIdIndexes[mergeTupleIndex], NullDatum.get());
+ if (tuples[i] != null) {
+ mergedTuple.put(mergeTupleIndex, tuples[i].get(j));
+ } else {
+ mergedTuple.put(mergeTupleIndex, NullDatum.get());
+ }
}
}
- mergeTupleIndex++;
+ resultColumnIdx++;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/95292d29/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
index 40ee54e..935e520 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
@@ -288,20 +288,84 @@ public class TestGroupByQuery extends QueryTestCaseBase {
schema.addColumn("code", Type.TEXT);
schema.addColumn("qty", Type.INT4);
schema.addColumn("qty2", Type.FLOAT8);
- String[] data = new String[]{ "1|a|3|3.0", "1|a|4|4.0", "1|b|5|5.0", "2|a|1|6.0", "2|c|2|7.0", "2|d|3|8.0" };
+ String[] data = new String[]{"1|a|3|3.0", "1|a|4|4.0", "1|b|5|5.0", "2|a|1|6.0", "2|c|2|7.0", "2|d|3|8.0"};
TajoTestingCluster.createTable("table10", schema, tableOptions, data);
res = executeString("select id, count(distinct code), " +
"avg(qty), min(qty), max(qty), sum(qty), " +
"cast(avg(qty2) as INT8), cast(min(qty2) as INT8), cast(max(qty2) as INT8), cast(sum(qty2) as INT8) " +
"from table10 group by id");
- String result = resultSetToString(res);
String expected = "id,?count_4,?avg_5,?min_6,?max_7,?sum_8,?cast_9,?cast_10,?cast_11,?cast_12\n" +
"-------------------------------\n" +
"1,2,4.0,0,5,12,4,0,5,12\n" +
"2,3,2.0,0,3,6,7,0,8,21\n";
+ assertEquals(expected, resultSetToString(res));
+
+ // multiple distinct with expression
+ res = executeString(
+ "select count(distinct code) + count(distinct qty) from table10"
+ );
+
+ expected = "?plus_2\n" +
+ "-------------------------------\n" +
+ "9\n";
+
+ assertEquals(expected, resultSetToString(res));
+ res.close();
+
+ res = executeString(
+ "select id, count(distinct code) + count(distinct qty) from table10 group by id"
+ );
+
+ expected = "id,?plus_2\n" +
+ "-------------------------------\n" +
+ "1,5\n" +
+ "2,6\n";
+
+ assertEquals(expected, resultSetToString(res));
+ res.close();
+
+ executeString("DROP TABLE table10 PURGE").close();
+ }
+
+ @Test
+ public final void testDistinctAggregationCasebyCase2() throws Exception {
+ // first distinct is smaller than second distinct.
+ KeyValueSet tableOptions = new KeyValueSet();
+ tableOptions.put(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.put(StorageConstants.CSVFILE_NULL, "\\\\N");
+
+ Schema schema = new Schema();
+ schema.addColumn("col1", Type.TEXT);
+ schema.addColumn("col2", Type.TEXT);
+ schema.addColumn("col3", Type.TEXT);
+
+ String[] data = new String[]{
+ "a|b-1|\\N",
+ "a|b-2|\\N",
+ "a|b-2|\\N",
+ "a|b-3|\\N",
+ "a|b-3|\\N",
+ "a|b-3|\\N"
+ };
+
+ TajoTestingCluster.createTable("table10", schema, tableOptions, data);
+
+ ResultSet res = executeString(
+ "select col1 \n" +
+ ",count(distinct col2) as cnt1\n" +
+ ",count(distinct case when col3 is not null then col2 else null end) as cnt2\n" +
+ "from table10 \n" +
+ "group by col1"
+ );
+ String result = resultSetToString(res);
+
+ String expected = "col1,cnt1,cnt2\n" +
+ "-------------------------------\n" +
+ "a,3,1\n";
+
assertEquals(expected, result);
executeString("DROP TABLE table10 PURGE").close();
http://git-wip-us.apache.org/repos/asf/tajo/blob/95292d29/tajo-storage/src/main/java/org/apache/tajo/storage/v2/RCFile.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/v2/RCFile.java b/tajo-storage/src/main/java/org/apache/tajo/storage/v2/RCFile.java
index 4b79c51..47dce74 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/v2/RCFile.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/v2/RCFile.java
@@ -1457,11 +1457,6 @@ public class RCFile {
currentKeyLength = sin.readInt();
compressedKeyLen = sin.readInt();
-// System.out.println(">>>currentRecordLength=" + currentRecordLength +
-// ",currentKeyLength=" + currentKeyLength +
-// ",compressedKeyLen=" + compressedKeyLen +
-// ",decompress=" + decompress);
-
if (decompress) {
keyTempBuffer.reset();
keyTempBuffer.write(sin, compressedKeyLen);
[10/19] git commit: Fixed a test failure by a non-deterministic query.
Posted by ji...@apache.org.
Fixed a test failure by a non-deterministic query.
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/9f873b12
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/9f873b12
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/9f873b12
Branch: refs/heads/index_support
Commit: 9f873b129de3ac0fc8122528b7e10e46406704d1
Parents: c87fe2e
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 20:05:48 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 20:05:48 2014 +0900
----------------------------------------------------------------------
.../test/java/org/apache/tajo/engine/query/TestGroupByQuery.java | 2 +-
.../queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql | 2 +-
.../results/TestGroupByQuery/testGroupByWithConstantKeys2.result | 4 ++--
3 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/9f873b12/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
index 935e520..41ffa06 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
@@ -149,7 +149,7 @@ public class TestGroupByQuery extends QueryTestCaseBase {
@Test
public final void testGroupByWithConstantKeys2() throws Exception {
// select l_partkey as a, timestamp '2014-07-07 04:28:31.561' as b, '##' as c, count(*) d from lineitem
- // group by a, b, c;
+ // group by a, b, c order by l_partkey;
ResultSet res = executeQuery();
assertResultSet(res);
cleanupQuery(res);
http://git-wip-us.apache.org/repos/asf/tajo/blob/9f873b12/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql
index 439d049..5a2892d 100644
--- a/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql
+++ b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupByWithConstantKeys2.sql
@@ -1 +1 @@
-select l_partkey as a, timestamp '2014-07-07 04:28:31.561' as b, '##' as c, count(*) d from lineitem group by a, b, c;
\ No newline at end of file
+select l_partkey as a, timestamp '2014-07-07 04:28:31.561' as b, '##' as c, count(*) d from lineitem group by a, b, c order by l_partkey;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/9f873b12/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result
index e2a3b89..776db3c 100644
--- a/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result
+++ b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupByWithConstantKeys2.result
@@ -1,5 +1,5 @@
a,b,c,d
-------------------------------
-3,2014-07-07 04:28:31.561,##,1
1,2014-07-07 04:28:31.561,##,2
-2,2014-07-07 04:28:31.561,##,2
\ No newline at end of file
+2,2014-07-07 04:28:31.561,##,2
+3,2014-07-07 04:28:31.561,##,1
\ No newline at end of file
[11/19] git commit: TAJO-908: Fetcher does not retry,
when pull server connection was closed. (jinho) Closes #58
Posted by ji...@apache.org.
TAJO-908: Fetcher does not retry, when pull server connection was closed. (jinho)
Closes #58
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/01857dad
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/01857dad
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/01857dad
Branch: refs/heads/index_support
Commit: 01857dadd42bd55894ce6edb55159e80c30f6c30
Parents: 9f873b1
Author: jinossy <ji...@gmail.com>
Authored: Fri Jul 11 22:07:43 2014 +0900
Committer: jinossy <ji...@gmail.com>
Committed: Fri Jul 11 22:07:43 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +
.../java/org/apache/tajo/conf/TajoConf.java | 3 +
tajo-common/src/main/proto/tajo_protos.proto | 1 +
.../java/org/apache/tajo/worker/Fetcher.java | 119 ++++++++++----
.../main/java/org/apache/tajo/worker/Task.java | 24 +--
.../org/apache/tajo/worker/TestFetcher.java | 162 ++++++++++++++-----
.../tajo/pullserver/TajoPullServerService.java | 26 ++-
7 files changed, 244 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/01857dad/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 2feba78..87cf7b0 100644
--- a/CHANGES
+++ b/CHANGES
@@ -81,6 +81,9 @@ Release 0.9.0 - unreleased
(Hyoungjun Kim via hyunsik)
BUG FIXES
+
+ TAJO-908: Fetcher does not retry, when pull server connection was closed.
+ (jinho)
TAJO-926: Join condition including column references of a row-preserving
table in left outer join causes incorrect result. (hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/01857dad/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 6298d27..dd5327d 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -217,6 +217,9 @@ public class TajoConf extends Configuration {
SHUFFLE_SSL_ENABLED_KEY("tajo.pullserver.ssl.enabled", false),
SHUFFLE_FILE_FORMAT("tajo.shuffle.file-format", "RAW"),
SHUFFLE_FETCHER_PARALLEL_EXECUTION_MAX_NUM("tajo.shuffle.fetcher.parallel-execution.max-num", 2),
+ SHUFFLE_FETCHER_CHUNK_MAX_SIZE("tajo.shuffle.fetcher.chunk.max-size", 8192 * 8),
+ SHUFFLE_FETCHER_READ_TIMEOUT("tajo.shuffle.fetcher.read.timeout-sec", 5),
+ SHUFFLE_FETCHER_READ_RETRY_MAX_NUM("tajo.shuffle.fetcher.read.retry.max-num", 5),
//////////////////////////////////
// Storage Configuration
http://git-wip-us.apache.org/repos/asf/tajo/blob/01857dad/tajo-common/src/main/proto/tajo_protos.proto
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/proto/tajo_protos.proto b/tajo-common/src/main/proto/tajo_protos.proto
index a7aa4f7..edd27fc 100644
--- a/tajo-common/src/main/proto/tajo_protos.proto
+++ b/tajo-common/src/main/proto/tajo_protos.proto
@@ -51,4 +51,5 @@ enum FetcherState {
FETCH_INIT = 0;
FETCH_FETCHING = 1;
FETCH_FINISHED = 2;
+ FETCH_FAILED = 3;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/01857dad/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java b/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
index 37c653c..2aa2875 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
@@ -22,11 +22,16 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.IOUtils;
import org.apache.tajo.TajoProtos;
+import org.apache.tajo.conf.TajoConf;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
import org.jboss.netty.handler.codec.http.*;
+import org.jboss.netty.handler.timeout.ReadTimeoutException;
+import org.jboss.netty.handler.timeout.ReadTimeoutHandler;
+import org.jboss.netty.util.HashedWheelTimer;
+import org.jboss.netty.util.Timer;
import java.io.File;
import java.io.FileNotFoundException;
@@ -48,6 +53,7 @@ public class Fetcher {
private final URI uri;
private final File file;
+ private final TajoConf conf;
private final String host;
private int port;
@@ -57,13 +63,15 @@ public class Fetcher {
private long fileLen;
private int messageReceiveCount;
private TajoProtos.FetcherState state;
+ private Timer timer;
private ClientBootstrap bootstrap;
- public Fetcher(URI uri, File file, ClientSocketChannelFactory factory) {
+ public Fetcher(TajoConf conf, URI uri, File file, ClientSocketChannelFactory factory) {
this.uri = uri;
this.file = file;
this.state = TajoProtos.FetcherState.FETCH_INIT;
+ this.conf = conf;
String scheme = uri.getScheme() == null ? "http" : uri.getScheme();
this.host = uri.getHost() == null ? "localhost" : uri.getHost();
@@ -106,41 +114,47 @@ public class Fetcher {
}
public File get() throws IOException {
- startTime = System.currentTimeMillis();
+ this.startTime = System.currentTimeMillis();
this.state = TajoProtos.FetcherState.FETCH_FETCHING;
- ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
+ try {
+ ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
- // Wait until the connection attempt succeeds or fails.
- Channel channel = future.awaitUninterruptibly().getChannel();
- if (!future.isSuccess()) {
- future.getChannel().close();
- throw new IOException(future.getCause());
- }
-
- String query = uri.getPath()
- + (uri.getRawQuery() != null ? "?" + uri.getRawQuery() : "");
- // Prepare the HTTP request.
- HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, query);
- request.setHeader(HttpHeaders.Names.HOST, host);
- LOG.info("Fetch: " + uri);
- request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
- request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
+ // Wait until the connection attempt succeeds or fails.
+ Channel channel = future.awaitUninterruptibly().getChannel();
+ if (!future.isSuccess()) {
+ future.getChannel().close();
+ state = TajoProtos.FetcherState.FETCH_FAILED;
+ throw new IOException(future.getCause());
+ }
+ String query = uri.getPath()
+ + (uri.getRawQuery() != null ? "?" + uri.getRawQuery() : "");
+ // Prepare the HTTP request.
+ HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, query);
+ request.setHeader(HttpHeaders.Names.HOST, host);
+ request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
+ request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
- // Send the HTTP request.
- ChannelFuture channelFuture = channel.write(request);
+ LOG.info("Status: " + getState() + ", URI:" + uri);
+ // Send the HTTP request.
+ ChannelFuture channelFuture = channel.write(request);
- // Wait for the server to close the connection.
- channel.getCloseFuture().awaitUninterruptibly();
+ // Wait for the server to close the connection.
+ channel.getCloseFuture().awaitUninterruptibly();
- channelFuture.addListener(ChannelFutureListener.CLOSE);
+ channelFuture.addListener(ChannelFutureListener.CLOSE);
- // Close the channel to exit.
- future.getChannel().close();
- finishTime = System.currentTimeMillis();
- this.state = TajoProtos.FetcherState.FETCH_FINISHED;
- return file;
+ // Close the channel to exit.
+ future.getChannel().close();
+ return file;
+ } finally {
+ this.finishTime = System.currentTimeMillis();
+ LOG.info("Status: " + getState() + ", URI:" + uri);
+ if (timer != null) {
+ timer.stop();
+ }
+ }
}
public URI getURI() {
@@ -161,9 +175,11 @@ public class Fetcher {
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
+
messageReceiveCount++;
try {
- if (!readingChunks) {
+ if (!readingChunks && e.getMessage() instanceof HttpResponse) {
+
HttpResponse response = (HttpResponse) e.getMessage();
StringBuilder sb = new StringBuilder();
@@ -188,8 +204,13 @@ public class Fetcher {
LOG.debug(sb.toString());
}
- if (response.getStatus() == HttpResponseStatus.NO_CONTENT) {
- LOG.info("There are no data corresponding to the request");
+ if (response.getStatus().getCode() == HttpResponseStatus.NO_CONTENT.getCode()) {
+ LOG.warn("There are no data corresponding to the request");
+ length = 0;
+ return;
+ } else if (response.getStatus().getCode() != HttpResponseStatus.OK.getCode()){
+ LOG.error(response.getStatus().getReasonPhrase());
+ state = TajoProtos.FetcherState.FETCH_FAILED;
return;
}
@@ -217,7 +238,9 @@ public class Fetcher {
+ "(received/total: " + fileLength + "/" + length + ")");
}
} else {
- fc.write(chunk.getContent().toByteBuffer());
+ if(fc != null){
+ fc.write(chunk.getContent().toByteBuffer());
+ }
}
}
} finally {
@@ -225,12 +248,32 @@ public class Fetcher {
fileLen = file.length();
}
- if(fileLen >= length){
+ if(fileLen == length){
IOUtils.cleanup(LOG, fc, raf);
-
+ finishTime = System.currentTimeMillis();
+ state = TajoProtos.FetcherState.FETCH_FINISHED;
}
}
}
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
+ throws Exception {
+ if (e.getCause() instanceof ReadTimeoutException) {
+ LOG.warn(e.getCause());
+ } else {
+ LOG.error("Fetch failed :", e.getCause());
+ }
+
+ if(ctx.getChannel().isConnected()){
+ ctx.getChannel().close().setFailure(e.getCause());
+ }
+
+ // this fetching will be retry
+ IOUtils.cleanup(LOG, fc, raf);
+ finishTime = System.currentTimeMillis();
+ state = TajoProtos.FetcherState.FETCH_FAILED;
+ }
}
class HttpClientPipelineFactory implements
@@ -245,8 +288,14 @@ public class Fetcher {
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = pipeline();
- pipeline.addLast("codec", new HttpClientCodec());
+ timer = new HashedWheelTimer();
+
+ int maxChunkSize = conf.getIntVar(TajoConf.ConfVars.SHUFFLE_FETCHER_CHUNK_MAX_SIZE);
+ int readTimeout = conf.getIntVar(TajoConf.ConfVars.SHUFFLE_FETCHER_READ_TIMEOUT);
+
+ pipeline.addLast("codec", new HttpClientCodec(4096, 8192, maxChunkSize));
pipeline.addLast("inflater", new HttpContentDecompressor());
+ pipeline.addLast("timeout", new ReadTimeoutHandler(timer, readTimeout));
pipeline.addLast("handler", new HttpClientHandler(file));
return pipeline;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/01857dad/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
index a960f69..9350838 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
@@ -576,23 +576,25 @@ public class Task {
private class FetchRunner implements Runnable {
private final TaskAttemptContext ctx;
private final Fetcher fetcher;
+ private int maxRetryNum;
public FetchRunner(TaskAttemptContext ctx, Fetcher fetcher) {
this.ctx = ctx;
this.fetcher = fetcher;
+ this.maxRetryNum = systemConf.getIntVar(TajoConf.ConfVars.SHUFFLE_FETCHER_READ_RETRY_MAX_NUM);
}
@Override
public void run() {
int retryNum = 0;
- int maxRetryNum = 5;
- int retryWaitTime = 1000;
+ int retryWaitTime = 1000; //sec
try { // for releasing fetch latch
while(!killed && retryNum < maxRetryNum) {
if (retryNum > 0) {
try {
Thread.sleep(retryWaitTime);
+ retryWaitTime = Math.min(10 * 1000, retryWaitTime * 2); // max 10 seconds
} catch (InterruptedException e) {
LOG.error(e);
}
@@ -600,7 +602,7 @@ public class Task {
}
try {
File fetched = fetcher.get();
- if (fetched != null) {
+ if (fetcher.getState() == TajoProtos.FetcherState.FETCH_FINISHED && fetched != null) {
break;
}
} catch (IOException e) {
@@ -609,11 +611,15 @@ public class Task {
retryNum++;
}
} finally {
- fetcherFinished(ctx);
- }
-
- if (retryNum == maxRetryNum) {
- LOG.error("ERROR: the maximum retry (" + retryNum + ") on the fetch exceeded (" + fetcher.getURI() + ")");
+ if(fetcher.getState() == TajoProtos.FetcherState.FETCH_FINISHED){
+ fetcherFinished(ctx);
+ } else {
+ if (retryNum == maxRetryNum) {
+ LOG.error("ERROR: the maximum retry (" + retryNum + ") on the fetch exceeded (" + fetcher.getURI() + ")");
+ }
+ aborted = true; // retry queryUnit
+ ctx.getFetchLatch().countDown();
+ }
}
}
}
@@ -674,7 +680,7 @@ public class Task {
storeDir.mkdirs();
}
storeFile = new File(storeDir, "in_" + i);
- Fetcher fetcher = new Fetcher(uri, storeFile, channelFactory);
+ Fetcher fetcher = new Fetcher(systemConf, uri, storeFile, channelFactory);
runnerList.add(fetcher);
i++;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/01857dad/tajo-core/src/test/java/org/apache/tajo/worker/TestFetcher.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/worker/TestFetcher.java b/tajo-core/src/test/java/org/apache/tajo/worker/TestFetcher.java
index c933294..82d662b 100644
--- a/tajo-core/src/test/java/org/apache/tajo/worker/TestFetcher.java
+++ b/tajo-core/src/test/java/org/apache/tajo/worker/TestFetcher.java
@@ -18,70 +18,87 @@
package org.apache.tajo.worker;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.fs.*;
+import org.apache.tajo.QueryId;
+import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.pullserver.TajoPullServerService;
import org.apache.tajo.rpc.RpcChannelFactory;
import org.apache.tajo.util.CommonTestingUtil;
-import org.apache.tajo.worker.dataserver.HttpDataServer;
-import org.apache.tajo.worker.dataserver.retriever.DataRetriever;
-import org.apache.tajo.worker.dataserver.retriever.DirectoryRetriever;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
+import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
-import java.io.FileWriter;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Random;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
public class TestFetcher {
private String TEST_DATA = "target/test-data/TestFetcher";
private String INPUT_DIR = TEST_DATA+"/in/";
private String OUTPUT_DIR = TEST_DATA+"/out/";
+ private TajoConf conf = new TajoConf();
+ private TajoPullServerService pullServerService;
+ private ClientSocketChannelFactory channelFactory;
@Before
public void setUp() throws Exception {
CommonTestingUtil.getTestDir(TEST_DATA);
CommonTestingUtil.getTestDir(INPUT_DIR);
CommonTestingUtil.getTestDir(OUTPUT_DIR);
+ conf.setVar(TajoConf.ConfVars.WORKER_TEMPORAL_DIR, INPUT_DIR);
+ conf.setIntVar(TajoConf.ConfVars.SHUFFLE_FETCHER_READ_TIMEOUT, 1);
+ conf.setIntVar(TajoConf.ConfVars.SHUFFLE_FETCHER_CHUNK_MAX_SIZE, 127);
+
+ pullServerService = new TajoPullServerService();
+ pullServerService.init(conf);
+ pullServerService.start();
+
+ channelFactory = RpcChannelFactory.createClientChannelFactory("Fetcher", 1);
+ }
+
+ @After
+ public void tearDown(){
+ pullServerService.stop();
+ channelFactory.releaseExternalResources();
}
@Test
public void testGet() throws IOException {
Random rnd = new Random();
- FileWriter writer = new FileWriter(INPUT_DIR + "data");
- String data;
+ QueryId queryId = QueryIdFactory.NULL_QUERY_ID;
+ String sid = "1";
+ String ta = "1_0";
+ String partId = "1";
+
+ String dataPath = INPUT_DIR + queryId.toString() + "/output"+ "/" + sid + "/" +ta + "/output/" + partId;
+ String params = String.format("qid=%s&sid=%s&p=%s&type=%s&ta=%s", queryId, sid, partId, "h", ta);
+
+ Path inputPath = new Path(dataPath);
+ FSDataOutputStream stream = LocalFileSystem.get(conf).create(inputPath, true);
for (int i = 0; i < 100; i++) {
- data = ""+rnd.nextInt();
- writer.write(data);
+ String data = ""+rnd.nextInt();
+ stream.write(data.getBytes());
}
- writer.flush();
- writer.close();
-
- DataRetriever ret = new DirectoryRetriever(INPUT_DIR);
- HttpDataServer server = new HttpDataServer(
- NetUtils.createSocketAddr("127.0.0.1:0"), ret);
- server.start();
- InetSocketAddress addr = server.getBindAddress();
-
- URI uri = URI.create("http://127.0.0.1:"+addr.getPort() + "/data");
- ClientSocketChannelFactory channelFactory = RpcChannelFactory.createClientChannelFactory("Fetcher", 1);
- Fetcher fetcher = new Fetcher(uri, new File(OUTPUT_DIR + "data"), channelFactory);
- fetcher.get();
- server.stop();
-
+ stream.flush();
+ stream.close();
+
+ URI uri = URI.create("http://127.0.0.1:" + pullServerService.getPort() + "/?" + params);
+ final Fetcher fetcher = new Fetcher(conf, uri, new File(OUTPUT_DIR + "data"), channelFactory);
+ assertNotNull(fetcher.get());
+
FileSystem fs = FileSystem.getLocal(new TajoConf());
- FileStatus inStatus = fs.getFileStatus(new Path(INPUT_DIR, "data"));
+ FileStatus inStatus = fs.getFileStatus(inputPath);
FileStatus outStatus = fs.getFileStatus(new Path(OUTPUT_DIR, "data"));
+
assertEquals(inStatus.getLen(), outStatus.getLen());
+ assertEquals(TajoProtos.FetcherState.FETCH_FINISHED, fetcher.getState());
}
@Test
@@ -96,29 +113,86 @@ public class TestFetcher {
@Test
public void testStatus() throws Exception {
Random rnd = new Random();
- FileWriter writer = new FileWriter(INPUT_DIR + "data");
- String data;
+ QueryId queryId = QueryIdFactory.NULL_QUERY_ID;
+ String sid = "1";
+ String ta = "1_0";
+ String partId = "1";
+
+ String dataPath = INPUT_DIR + queryId.toString() + "/output"+ "/" + sid + "/" +ta + "/output/" + partId;
+ String params = String.format("qid=%s&sid=%s&p=%s&type=%s&ta=%s", queryId, sid, partId, "h", ta);
+
+ FSDataOutputStream stream = LocalFileSystem.get(conf).create(new Path(dataPath), true);
for (int i = 0; i < 100; i++) {
- data = ""+rnd.nextInt();
- writer.write(data);
+ String data = ""+rnd.nextInt();
+ stream.write(data.getBytes());
}
- writer.flush();
- writer.close();
+ stream.flush();
+ stream.close();
+
+ URI uri = URI.create("http://127.0.0.1:" + pullServerService.getPort() + "/?" + params);
+ final Fetcher fetcher = new Fetcher(conf, uri, new File(OUTPUT_DIR + "data"), channelFactory);
+ assertEquals(TajoProtos.FetcherState.FETCH_INIT, fetcher.getState());
- DataRetriever ret = new DirectoryRetriever(INPUT_DIR);
- final HttpDataServer server = new HttpDataServer(
- NetUtils.createSocketAddr("127.0.0.1:0"), ret);
- server.start();
- InetSocketAddress addr = server.getBindAddress();
+ fetcher.get();
+ assertEquals(TajoProtos.FetcherState.FETCH_FINISHED, fetcher.getState());
+ }
+
+ @Test
+ public void testNoContentFetch() throws Exception {
- URI uri = URI.create("http://127.0.0.1:"+addr.getPort() + "/data");
- ClientSocketChannelFactory channelFactory = RpcChannelFactory.createClientChannelFactory("Fetcher", 1);
+ QueryId queryId = QueryIdFactory.NULL_QUERY_ID;
+ String sid = "1";
+ String ta = "1_0";
+ String partId = "1";
- final Fetcher fetcher = new Fetcher(uri, new File(OUTPUT_DIR + "data"), channelFactory);
+ String dataPath = INPUT_DIR + queryId.toString() + "/output"+ "/" + sid + "/" +ta + "/output/" + partId;
+ String params = String.format("qid=%s&sid=%s&p=%s&type=%s&ta=%s", queryId, sid, partId, "h", ta);
+
+ Path inputPath = new Path(dataPath);
+ if(LocalFileSystem.get(conf).exists(inputPath)){
+ LocalFileSystem.get(conf).delete(new Path(dataPath), true);
+ }
+
+ FSDataOutputStream stream = LocalFileSystem.get(conf).create(new Path(dataPath).getParent(), true);
+ stream.close();
+
+ URI uri = URI.create("http://127.0.0.1:" + pullServerService.getPort() + "/?" + params);
+ final Fetcher fetcher = new Fetcher(conf, uri, new File(OUTPUT_DIR + "data"), channelFactory);
assertEquals(TajoProtos.FetcherState.FETCH_INIT, fetcher.getState());
fetcher.get();
assertEquals(TajoProtos.FetcherState.FETCH_FINISHED, fetcher.getState());
- server.stop();
+ }
+
+ @Test
+ public void testFailureStatus() throws Exception {
+ Random rnd = new Random();
+
+ QueryId queryId = QueryIdFactory.NULL_QUERY_ID;
+ String sid = "1";
+ String ta = "1_0";
+ String partId = "1";
+
+ String dataPath = INPUT_DIR + queryId.toString() + "/output"+ "/" + sid + "/" +ta + "/output/" + partId;
+
+ //TajoPullServerService will be throws BAD_REQUEST by Unknown shuffle type
+ String shuffleType = "x";
+ String params = String.format("qid=%s&sid=%s&p=%s&type=%s&ta=%s", queryId, sid, partId, shuffleType, ta);
+
+ FSDataOutputStream stream = LocalFileSystem.get(conf).create(new Path(dataPath), true);
+
+ for (int i = 0; i < 100; i++) {
+ String data = params + rnd.nextInt();
+ stream.write(data.getBytes());
+ }
+ stream.flush();
+ stream.close();
+
+ URI uri = URI.create("http://127.0.0.1:" + pullServerService.getPort() + "/?" + params);
+ final Fetcher fetcher = new Fetcher(conf, uri, new File(OUTPUT_DIR + "data"), channelFactory);
+ assertEquals(TajoProtos.FetcherState.FETCH_INIT, fetcher.getState());
+
+ fetcher.get();
+ assertEquals(TajoProtos.FetcherState.FETCH_FAILED, fetcher.getState());
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/01857dad/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
----------------------------------------------------------------------
diff --git a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
index cc3cb2e..373642b 100644
--- a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
+++ b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
@@ -299,9 +299,11 @@ public class TajoPullServerService extends AbstractService {
if (sslFactory != null) {
pipeline.addLast("ssl", new SslHandler(sslFactory.createSSLEngine()));
}
- pipeline.addLast("decoder", new HttpRequestDecoder());
+
+ int maxChunkSize = getConfig().getInt(ConfVars.SHUFFLE_FETCHER_CHUNK_MAX_SIZE.varname,
+ ConfVars.SHUFFLE_FETCHER_CHUNK_MAX_SIZE.defaultIntVal);
+ pipeline.addLast("codec", new HttpServerCodec(4096, 8192, maxChunkSize));
pipeline.addLast("aggregator", new HttpChunkAggregator(1 << 16));
- pipeline.addLast("encoder", new HttpResponseEncoder());
pipeline.addLast("chunking", new ChunkedWriteHandler());
pipeline.addLast("shuffle", PullServer);
return pipeline;
@@ -319,11 +321,14 @@ public class TajoPullServerService extends AbstractService {
new LocalDirAllocator(ConfVars.WORKER_TEMPORAL_DIR.varname);
private int port;
- public PullServer(Configuration conf) {
+ public PullServer(Configuration conf) throws IOException {
this.conf = conf;
// indexCache = new IndexCache(new JobConf(conf));
this.port = conf.getInt(ConfVars.PULLSERVER_PORT.varname,
ConfVars.PULLSERVER_PORT.defaultIntVal);
+
+ // init local temporal dir
+ lDirAlloc.getAllLocalPathsToRead(".", conf);
}
public void setPort(int port) {
@@ -402,9 +407,13 @@ public class TajoPullServerService extends AbstractService {
// if a subquery requires a range shuffle
if (shuffleType.equals("r")) {
String ta = taskIds.get(0);
+ if(!lDirAlloc.ifExists(queryBaseDir + "/" + sid + "/" + ta + "/output/", conf)){
+ LOG.warn(e);
+ sendError(ctx, NO_CONTENT);
+ return;
+ }
Path path = localFS.makeQualified(
lDirAlloc.getLocalPathToRead(queryBaseDir + "/" + sid + "/" + ta + "/output/", conf));
-
String startKey = params.get("start").get(0);
String endKey = params.get("end").get(0);
boolean last = params.get("final") != null;
@@ -424,15 +433,20 @@ public class TajoPullServerService extends AbstractService {
// if a subquery requires a hash shuffle
} else if (shuffleType.equals("h")) {
for (String ta : taskIds) {
+ if (!lDirAlloc.ifExists(queryBaseDir + "/" + sid + "/" + ta + "/output/" + partId, conf)) {
+ LOG.warn(e);
+ sendError(ctx, NO_CONTENT);
+ return;
+ }
Path path = localFS.makeQualified(
- lDirAlloc.getLocalPathToRead(queryBaseDir + "/" + sid + "/" +
- ta + "/output/" + partId, conf));
+ lDirAlloc.getLocalPathToRead(queryBaseDir + "/" + sid + "/" + ta + "/output/" + partId, conf));
File file = new File(path.toUri());
FileChunk chunk = new FileChunk(file, 0, file.length());
chunks.add(chunk);
}
} else {
LOG.error("Unknown shuffle type: " + shuffleType);
+ sendError(ctx, "Unknown shuffle type:" + shuffleType, BAD_REQUEST);
return;
}
[18/19] git commit: TAJO-933: Fork some classes of Parquet as builtin
third-party classes.
Posted by ji...@apache.org.
TAJO-933: Fork some classes of Parquet as builtin third-party classes.
Closes #75
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/9cf10719
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/9cf10719
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/9cf10719
Branch: refs/heads/index_support
Commit: 9cf107191bac3592d55c59ae6b60ab6051d5656d
Parents: 47d4fe2
Author: Hyunsik Choi <hy...@apache.org>
Authored: Tue Jul 15 15:21:51 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Tue Jul 15 15:21:51 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +
.../tajo/storage/parquet/TajoParquetReader.java | 2 +-
.../tajo/storage/parquet/TajoParquetWriter.java | 2 +-
.../thirdparty/parquet/CodecFactory.java | 196 ++++++++
.../parquet/ColumnChunkPageWriteStore.java | 211 ++++++++
.../parquet/InternalParquetRecordReader.java | 187 +++++++
.../parquet/InternalParquetRecordWriter.java | 156 ++++++
.../thirdparty/parquet/ParquetFileWriter.java | 504 +++++++++++++++++++
.../thirdparty/parquet/ParquetReader.java | 151 ++++++
.../thirdparty/parquet/ParquetWriter.java | 220 ++++++++
10 files changed, 1630 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index d819d20..4aeb20c 100644
--- a/CHANGES
+++ b/CHANGES
@@ -27,6 +27,9 @@ Release 0.9.0 - unreleased
IMPROVEMENT
+ TAJO-933: Fork some classes of Parquet as builtin third-party classes.
+ (hyunsik)
+
TAJO-932: Upgrade Parquet to 1.5.0. (hyunsik)
TAJO-903: Some left outer join cases are not optimized as the broadcast
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetReader.java b/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetReader.java
index 660a398..a765f48 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetReader.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetReader.java
@@ -21,8 +21,8 @@ package org.apache.tajo.storage.parquet;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.thirdparty.parquet.ParquetReader;
import parquet.filter.UnboundRecordFilter;
-import parquet.hadoop.ParquetReader;
import java.io.IOException;
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetWriter.java b/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetWriter.java
index 5e31782..69b76c4 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetWriter.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoParquetWriter.java
@@ -21,7 +21,7 @@ package org.apache.tajo.storage.parquet;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.storage.Tuple;
-import parquet.hadoop.ParquetWriter;
+import org.apache.tajo.storage.thirdparty.parquet.ParquetWriter;
import parquet.hadoop.metadata.CompressionCodecName;
import java.io.IOException;
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/CodecFactory.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/CodecFactory.java b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/CodecFactory.java
new file mode 100644
index 0000000..543336f
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/CodecFactory.java
@@ -0,0 +1,196 @@
+/**
+ * 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.storage.thirdparty.parquet;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.CodecPool;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionOutputStream;
+import org.apache.hadoop.io.compress.Compressor;
+import org.apache.hadoop.io.compress.Decompressor;
+import org.apache.hadoop.util.ReflectionUtils;
+
+import parquet.bytes.BytesInput;
+import parquet.hadoop.BadConfigurationException;
+import parquet.hadoop.metadata.CompressionCodecName;
+
+class CodecFactory {
+
+ public class BytesDecompressor {
+
+ private final CompressionCodec codec;
+ private final Decompressor decompressor;
+
+ public BytesDecompressor(CompressionCodec codec) {
+ this.codec = codec;
+ if (codec != null) {
+ decompressor = CodecPool.getDecompressor(codec);
+ } else {
+ decompressor = null;
+ }
+ }
+
+ public BytesInput decompress(BytesInput bytes, int uncompressedSize) throws IOException {
+ final BytesInput decompressed;
+ if (codec != null) {
+ decompressor.reset();
+ InputStream is = codec.createInputStream(new ByteArrayInputStream(bytes.toByteArray()), decompressor);
+ decompressed = BytesInput.from(is, uncompressedSize);
+ } else {
+ decompressed = bytes;
+ }
+ return decompressed;
+ }
+
+ private void release() {
+ if (decompressor != null) {
+ CodecPool.returnDecompressor(decompressor);
+ }
+ }
+ }
+
+ /**
+ * Encapsulates the logic around hadoop compression
+ *
+ * @author Julien Le Dem
+ *
+ */
+ public static class BytesCompressor {
+
+ private final CompressionCodec codec;
+ private final Compressor compressor;
+ private final ByteArrayOutputStream compressedOutBuffer;
+ private final CompressionCodecName codecName;
+
+ public BytesCompressor(CompressionCodecName codecName, CompressionCodec codec, int pageSize) {
+ this.codecName = codecName;
+ this.codec = codec;
+ if (codec != null) {
+ this.compressor = CodecPool.getCompressor(codec);
+ this.compressedOutBuffer = new ByteArrayOutputStream(pageSize);
+ } else {
+ this.compressor = null;
+ this.compressedOutBuffer = null;
+ }
+ }
+
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ final BytesInput compressedBytes;
+ if (codec == null) {
+ compressedBytes = bytes;
+ } else {
+ compressedOutBuffer.reset();
+ if (compressor != null) {
+ // null compressor for non-native gzip
+ compressor.reset();
+ }
+ CompressionOutputStream cos = codec.createOutputStream(compressedOutBuffer, compressor);
+ bytes.writeAllTo(cos);
+ cos.finish();
+ cos.close();
+ compressedBytes = BytesInput.from(compressedOutBuffer);
+ }
+ return compressedBytes;
+ }
+
+ private void release() {
+ if (compressor != null) {
+ CodecPool.returnCompressor(compressor);
+ }
+ }
+
+ public CompressionCodecName getCodecName() {
+ return codecName;
+ }
+
+ }
+
+ private final Map<CompressionCodecName, BytesCompressor> compressors = new HashMap<CompressionCodecName, BytesCompressor>();
+ private final Map<CompressionCodecName, BytesDecompressor> decompressors = new HashMap<CompressionCodecName, BytesDecompressor>();
+ private final Map<String, CompressionCodec> codecByName = new HashMap<String, CompressionCodec>();
+ private final Configuration configuration;
+
+ public CodecFactory(Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ /**
+ *
+ * @param codecName the requested codec
+ * @return the corresponding hadoop codec. null if UNCOMPRESSED
+ */
+ private CompressionCodec getCodec(CompressionCodecName codecName) {
+ String codecClassName = codecName.getHadoopCompressionCodecClassName();
+ if (codecClassName == null) {
+ return null;
+ }
+ CompressionCodec codec = codecByName.get(codecClassName);
+ if (codec != null) {
+ return codec;
+ }
+
+ try {
+ Class<?> codecClass = Class.forName(codecClassName);
+ codec = (CompressionCodec)ReflectionUtils.newInstance(codecClass, configuration);
+ codecByName.put(codecClassName, codec);
+ return codec;
+ } catch (ClassNotFoundException e) {
+ throw new BadConfigurationException("Class " + codecClassName + " was not found", e);
+ }
+ }
+
+ public BytesCompressor getCompressor(CompressionCodecName codecName, int pageSize) {
+ BytesCompressor comp = compressors.get(codecName);
+ if (comp == null) {
+ CompressionCodec codec = getCodec(codecName);
+ comp = new BytesCompressor(codecName, codec, pageSize);
+ compressors.put(codecName, comp);
+ }
+ return comp;
+ }
+
+ public BytesDecompressor getDecompressor(CompressionCodecName codecName) {
+ BytesDecompressor decomp = decompressors.get(codecName);
+ if (decomp == null) {
+ CompressionCodec codec = getCodec(codecName);
+ decomp = new BytesDecompressor(codec);
+ decompressors.put(codecName, decomp);
+ }
+ return decomp;
+ }
+
+ public void release() {
+ for (BytesCompressor compressor : compressors.values()) {
+ compressor.release();
+ }
+ compressors.clear();
+ for (BytesDecompressor decompressor : decompressors.values()) {
+ decompressor.release();
+ }
+ decompressors.clear();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ColumnChunkPageWriteStore.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ColumnChunkPageWriteStore.java b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ColumnChunkPageWriteStore.java
new file mode 100644
index 0000000..5f89ead
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ColumnChunkPageWriteStore.java
@@ -0,0 +1,211 @@
+/**
+ * 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.storage.thirdparty.parquet;
+
+import static org.apache.tajo.storage.thirdparty.parquet.CodecFactory.BytesCompressor;
+import static parquet.Log.INFO;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import parquet.Log;
+import parquet.bytes.BytesInput;
+import parquet.bytes.CapacityByteArrayOutputStream;
+import parquet.column.ColumnDescriptor;
+import parquet.column.Encoding;
+import parquet.column.page.DictionaryPage;
+import parquet.column.page.PageWriteStore;
+import parquet.column.page.PageWriter;
+import parquet.column.statistics.Statistics;
+import parquet.column.statistics.BooleanStatistics;
+import parquet.format.converter.ParquetMetadataConverter;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.MessageType;
+
+class ColumnChunkPageWriteStore implements PageWriteStore {
+ private static final Log LOG = Log.getLog(ColumnChunkPageWriteStore.class);
+
+ private static ParquetMetadataConverter parquetMetadataConverter = new ParquetMetadataConverter();
+
+ private static final class ColumnChunkPageWriter implements PageWriter {
+
+ private final ColumnDescriptor path;
+ private final BytesCompressor compressor;
+
+ private final CapacityByteArrayOutputStream buf;
+ private DictionaryPage dictionaryPage;
+
+ private long uncompressedLength;
+ private long compressedLength;
+ private long totalValueCount;
+ private int pageCount;
+
+ private Set<Encoding> encodings = new HashSet<Encoding>();
+
+ private Statistics totalStatistics;
+
+ private ColumnChunkPageWriter(ColumnDescriptor path, BytesCompressor compressor, int initialSize) {
+ this.path = path;
+ this.compressor = compressor;
+ this.buf = new CapacityByteArrayOutputStream(initialSize);
+ this.totalStatistics = Statistics.getStatsBasedOnType(this.path.getType());
+ }
+
+ @Deprecated
+ @Override
+ public void writePage(BytesInput bytes,
+ int valueCount,
+ Encoding rlEncoding,
+ Encoding dlEncoding,
+ Encoding valuesEncoding) throws IOException {
+ long uncompressedSize = bytes.size();
+ BytesInput compressedBytes = compressor.compress(bytes);
+ long compressedSize = compressedBytes.size();
+ BooleanStatistics statistics = new BooleanStatistics(); // dummy stats object
+ parquetMetadataConverter.writeDataPageHeader(
+ (int)uncompressedSize,
+ (int)compressedSize,
+ valueCount,
+ statistics,
+ rlEncoding,
+ dlEncoding,
+ valuesEncoding,
+ buf);
+ this.uncompressedLength += uncompressedSize;
+ this.compressedLength += compressedSize;
+ this.totalValueCount += valueCount;
+ this.pageCount += 1;
+ compressedBytes.writeAllTo(buf);
+ encodings.add(rlEncoding);
+ encodings.add(dlEncoding);
+ encodings.add(valuesEncoding);
+ }
+
+ @Override
+ public void writePage(BytesInput bytes,
+ int valueCount,
+ Statistics statistics,
+ Encoding rlEncoding,
+ Encoding dlEncoding,
+ Encoding valuesEncoding) throws IOException {
+ long uncompressedSize = bytes.size();
+ BytesInput compressedBytes = compressor.compress(bytes);
+ long compressedSize = compressedBytes.size();
+ parquetMetadataConverter.writeDataPageHeader(
+ (int)uncompressedSize,
+ (int)compressedSize,
+ valueCount,
+ statistics,
+ rlEncoding,
+ dlEncoding,
+ valuesEncoding,
+ buf);
+ this.uncompressedLength += uncompressedSize;
+ this.compressedLength += compressedSize;
+ this.totalValueCount += valueCount;
+ this.pageCount += 1;
+ this.totalStatistics.mergeStatistics(statistics);
+ compressedBytes.writeAllTo(buf);
+ encodings.add(rlEncoding);
+ encodings.add(dlEncoding);
+ encodings.add(valuesEncoding);
+ }
+
+ @Override
+ public long getMemSize() {
+ return buf.size();
+ }
+
+ public void writeToFileWriter(ParquetFileWriter writer) throws IOException {
+ writer.startColumn(path, totalValueCount, compressor.getCodecName());
+ if (dictionaryPage != null) {
+ writer.writeDictionaryPage(dictionaryPage);
+ encodings.add(dictionaryPage.getEncoding());
+ }
+ writer.writeDataPages(BytesInput.from(buf), uncompressedLength, compressedLength, totalStatistics, new ArrayList<Encoding>(encodings));
+ writer.endColumn();
+ if (INFO) {
+ LOG.info(
+ String.format(
+ "written %,dB for %s: %,d values, %,dB raw, %,dB comp, %d pages, encodings: %s",
+ buf.size(), path, totalValueCount, uncompressedLength, compressedLength, pageCount, encodings)
+ + (dictionaryPage != null ? String.format(
+ ", dic { %,d entries, %,dB raw, %,dB comp}",
+ dictionaryPage.getDictionarySize(), dictionaryPage.getUncompressedSize(), dictionaryPage.getDictionarySize())
+ : ""));
+ }
+ encodings.clear();
+ pageCount = 0;
+ }
+
+ @Override
+ public long allocatedSize() {
+ return buf.getCapacity();
+ }
+
+ @Override
+ public void writeDictionaryPage(DictionaryPage dictionaryPage) throws IOException {
+ if (this.dictionaryPage != null) {
+ throw new ParquetEncodingException("Only one dictionary page is allowed");
+ }
+ BytesInput dictionaryBytes = dictionaryPage.getBytes();
+ int uncompressedSize = (int)dictionaryBytes.size();
+ BytesInput compressedBytes = compressor.compress(dictionaryBytes);
+ this.dictionaryPage = new DictionaryPage(BytesInput.copy(compressedBytes), uncompressedSize, dictionaryPage.getDictionarySize(), dictionaryPage.getEncoding());
+ }
+
+ @Override
+ public String memUsageString(String prefix) {
+ return buf.memUsageString(prefix + " ColumnChunkPageWriter");
+ }
+ }
+
+ private final Map<ColumnDescriptor, ColumnChunkPageWriter> writers = new HashMap<ColumnDescriptor, ColumnChunkPageWriter>();
+ private final MessageType schema;
+ private final BytesCompressor compressor;
+ private final int initialSize;
+
+ public ColumnChunkPageWriteStore(BytesCompressor compressor, MessageType schema, int initialSize) {
+ this.compressor = compressor;
+ this.schema = schema;
+ this.initialSize = initialSize;
+ }
+
+ @Override
+ public PageWriter getPageWriter(ColumnDescriptor path) {
+ if (!writers.containsKey(path)) {
+ writers.put(path, new ColumnChunkPageWriter(path, compressor, initialSize));
+ }
+ return writers.get(path);
+ }
+
+ public void flushToFileWriter(ParquetFileWriter writer) throws IOException {
+ List<ColumnDescriptor> columns = schema.getColumns();
+ for (ColumnDescriptor columnDescriptor : columns) {
+ ColumnChunkPageWriter pageWriter = writers.get(columnDescriptor);
+ pageWriter.writeToFileWriter(writer);
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordReader.java b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordReader.java
new file mode 100644
index 0000000..61567e5
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordReader.java
@@ -0,0 +1,187 @@
+/**
+ * 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.storage.thirdparty.parquet;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import parquet.Log;
+import parquet.column.ColumnDescriptor;
+import parquet.column.page.PageReadStore;
+import parquet.filter.UnboundRecordFilter;
+import parquet.hadoop.ParquetFileReader;
+import parquet.hadoop.api.ReadSupport;
+import parquet.hadoop.metadata.BlockMetaData;
+import parquet.hadoop.util.counters.BenchmarkCounter;
+import parquet.io.ColumnIOFactory;
+import parquet.io.MessageColumnIO;
+import parquet.io.ParquetDecodingException;
+import parquet.io.api.RecordMaterializer;
+import parquet.schema.GroupType;
+import parquet.schema.MessageType;
+import parquet.schema.Type;
+
+import static java.lang.String.format;
+import static parquet.Log.DEBUG;
+
+class InternalParquetRecordReader<T> {
+ private static final Log LOG = Log.getLog(InternalParquetRecordReader.class);
+
+ private final ColumnIOFactory columnIOFactory = new ColumnIOFactory();
+
+ private MessageType requestedSchema;
+ private MessageType fileSchema;
+ private int columnCount;
+ private final ReadSupport<T> readSupport;
+
+ private RecordMaterializer<T> recordConverter;
+
+ private T currentValue;
+ private long total;
+ private int current = 0;
+ private int currentBlock = -1;
+ private ParquetFileReader reader;
+ private parquet.io.RecordReader<T> recordReader;
+ private UnboundRecordFilter recordFilter;
+
+ private long totalTimeSpentReadingBytes;
+ private long totalTimeSpentProcessingRecords;
+ private long startedAssemblingCurrentBlockAt;
+
+ private long totalCountLoadedSoFar = 0;
+
+ private Path file;
+
+ /**
+ * @param readSupport Object which helps reads files of the given type, e.g. Thrift, Avro.
+ */
+ public InternalParquetRecordReader(ReadSupport<T> readSupport) {
+ this(readSupport, null);
+ }
+
+ /**
+ * @param readSupport Object which helps reads files of the given type, e.g. Thrift, Avro.
+ * @param filter Optional filter for only returning matching records.
+ */
+ public InternalParquetRecordReader(ReadSupport<T> readSupport, UnboundRecordFilter
+ filter) {
+ this.readSupport = readSupport;
+ this.recordFilter = filter;
+ }
+
+ private void checkRead() throws IOException {
+ if (current == totalCountLoadedSoFar) {
+ if (current != 0) {
+ long timeAssembling = System.currentTimeMillis() - startedAssemblingCurrentBlockAt;
+ totalTimeSpentProcessingRecords += timeAssembling;
+ LOG.info("Assembled and processed " + totalCountLoadedSoFar + " records from " + columnCount + " columns in " + totalTimeSpentProcessingRecords + " ms: "+((float)totalCountLoadedSoFar / totalTimeSpentProcessingRecords) + " rec/ms, " + ((float)totalCountLoadedSoFar * columnCount / totalTimeSpentProcessingRecords) + " cell/ms");
+ long totalTime = totalTimeSpentProcessingRecords + totalTimeSpentReadingBytes;
+ long percentReading = 100 * totalTimeSpentReadingBytes / totalTime;
+ long percentProcessing = 100 * totalTimeSpentProcessingRecords / totalTime;
+ LOG.info("time spent so far " + percentReading + "% reading ("+totalTimeSpentReadingBytes+" ms) and " + percentProcessing + "% processing ("+totalTimeSpentProcessingRecords+" ms)");
+ }
+
+ LOG.info("at row " + current + ". reading next block");
+ long t0 = System.currentTimeMillis();
+ PageReadStore pages = reader.readNextRowGroup();
+ if (pages == null) {
+ throw new IOException("expecting more rows but reached last block. Read " + current + " out of " + total);
+ }
+ long timeSpentReading = System.currentTimeMillis() - t0;
+ totalTimeSpentReadingBytes += timeSpentReading;
+ BenchmarkCounter.incrementTime(timeSpentReading);
+ LOG.info("block read in memory in " + timeSpentReading + " ms. row count = " + pages.getRowCount());
+ if (Log.DEBUG) LOG.debug("initializing Record assembly with requested schema " + requestedSchema);
+ MessageColumnIO columnIO = columnIOFactory.getColumnIO(requestedSchema, fileSchema);
+ recordReader = columnIO.getRecordReader(pages, recordConverter, recordFilter);
+ startedAssemblingCurrentBlockAt = System.currentTimeMillis();
+ totalCountLoadedSoFar += pages.getRowCount();
+ ++ currentBlock;
+ }
+ }
+
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ public Void getCurrentKey() throws IOException, InterruptedException {
+ return null;
+ }
+
+ public T getCurrentValue() throws IOException,
+ InterruptedException {
+ return currentValue;
+ }
+
+ public float getProgress() throws IOException, InterruptedException {
+ return (float) current / total;
+ }
+
+ public void initialize(MessageType requestedSchema, MessageType fileSchema,
+ Map<String, String> extraMetadata, Map<String, String> readSupportMetadata,
+ Path file, List<BlockMetaData> blocks, Configuration configuration)
+ throws IOException {
+ this.requestedSchema = requestedSchema;
+ this.fileSchema = fileSchema;
+ this.file = file;
+ this.columnCount = this.requestedSchema.getPaths().size();
+ this.recordConverter = readSupport.prepareForRead(
+ configuration, extraMetadata, fileSchema,
+ new ReadSupport.ReadContext(requestedSchema, readSupportMetadata));
+
+ List<ColumnDescriptor> columns = requestedSchema.getColumns();
+ reader = new ParquetFileReader(configuration, file, blocks, columns);
+ for (BlockMetaData block : blocks) {
+ total += block.getRowCount();
+ }
+ LOG.info("RecordReader initialized will read a total of " + total + " records.");
+ }
+
+ private boolean contains(GroupType group, String[] path, int index) {
+ if (index == path.length) {
+ return false;
+ }
+ if (group.containsField(path[index])) {
+ Type type = group.getType(path[index]);
+ if (type.isPrimitive()) {
+ return index + 1 == path.length;
+ } else {
+ return contains(type.asGroupType(), path, index + 1);
+ }
+ }
+ return false;
+ }
+
+ public boolean nextKeyValue() throws IOException, InterruptedException {
+ if (current < total) {
+ try {
+ checkRead();
+ currentValue = recordReader.read();
+ if (DEBUG) LOG.debug("read value: " + currentValue);
+ current ++;
+ } catch (RuntimeException e) {
+ throw new ParquetDecodingException(format("Can not read value at %d in block %d in file %s", current, currentBlock, file), e);
+ }
+ return true;
+ }
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordWriter.java b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordWriter.java
new file mode 100644
index 0000000..8ce4b1c
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/InternalParquetRecordWriter.java
@@ -0,0 +1,156 @@
+/**
+ * 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.storage.thirdparty.parquet;
+
+import static java.lang.Math.max;
+import static java.lang.Math.min;
+import static java.lang.String.format;
+import static org.apache.tajo.storage.thirdparty.parquet.CodecFactory.BytesCompressor;
+import static parquet.Log.DEBUG;
+import static parquet.Preconditions.checkNotNull;
+
+import java.io.IOException;
+import java.util.Map;
+
+import parquet.Log;
+import parquet.column.ParquetProperties.WriterVersion;
+import parquet.column.impl.ColumnWriteStoreImpl;
+import parquet.hadoop.api.WriteSupport;
+import parquet.io.ColumnIOFactory;
+import parquet.io.MessageColumnIO;
+import parquet.schema.MessageType;
+
+class InternalParquetRecordWriter<T> {
+ private static final Log LOG = Log.getLog(InternalParquetRecordWriter.class);
+
+ private static final int MINIMUM_BUFFER_SIZE = 64 * 1024;
+ private static final int MINIMUM_RECORD_COUNT_FOR_CHECK = 100;
+ private static final int MAXIMUM_RECORD_COUNT_FOR_CHECK = 10000;
+
+ private final ParquetFileWriter w;
+ private final WriteSupport<T> writeSupport;
+ private final MessageType schema;
+ private final Map<String, String> extraMetaData;
+ private final int blockSize;
+ private final int pageSize;
+ private final BytesCompressor compressor;
+ private final int dictionaryPageSize;
+ private final boolean enableDictionary;
+ private final boolean validating;
+ private final WriterVersion writerVersion;
+
+ private long recordCount = 0;
+ private long recordCountForNextMemCheck = MINIMUM_RECORD_COUNT_FOR_CHECK;
+
+ private ColumnWriteStoreImpl store;
+ private ColumnChunkPageWriteStore pageStore;
+
+ /**
+ * @param w the file to write to
+ * @param writeSupport the class to convert incoming records
+ * @param schema the schema of the records
+ * @param extraMetaData extra meta data to write in the footer of the file
+ * @param blockSize the size of a block in the file (this will be approximate)
+ * @param codec the codec used to compress
+ */
+ public InternalParquetRecordWriter(
+ ParquetFileWriter w,
+ WriteSupport<T> writeSupport,
+ MessageType schema,
+ Map<String, String> extraMetaData,
+ int blockSize,
+ int pageSize,
+ BytesCompressor compressor,
+ int dictionaryPageSize,
+ boolean enableDictionary,
+ boolean validating,
+ WriterVersion writerVersion) {
+ this.w = w;
+ this.writeSupport = checkNotNull(writeSupport, "writeSupport");
+ this.schema = schema;
+ this.extraMetaData = extraMetaData;
+ this.blockSize = blockSize;
+ this.pageSize = pageSize;
+ this.compressor = compressor;
+ this.dictionaryPageSize = dictionaryPageSize;
+ this.enableDictionary = enableDictionary;
+ this.validating = validating;
+ this.writerVersion = writerVersion;
+ initStore();
+ }
+
+ private void initStore() {
+ // we don't want this number to be too small
+ // ideally we divide the block equally across the columns
+ // it is unlikely all columns are going to be the same size.
+ int initialBlockBufferSize = max(MINIMUM_BUFFER_SIZE, blockSize / schema.getColumns().size() / 5);
+ pageStore = new ColumnChunkPageWriteStore(compressor, schema, initialBlockBufferSize);
+ // we don't want this number to be too small either
+ // ideally, slightly bigger than the page size, but not bigger than the block buffer
+ int initialPageBufferSize = max(MINIMUM_BUFFER_SIZE, min(pageSize + pageSize / 10, initialBlockBufferSize));
+ store = new ColumnWriteStoreImpl(pageStore, pageSize, initialPageBufferSize, dictionaryPageSize, enableDictionary, writerVersion);
+ MessageColumnIO columnIO = new ColumnIOFactory(validating).getColumnIO(schema);
+ writeSupport.prepareForWrite(columnIO.getRecordWriter(store));
+ }
+
+ public void close() throws IOException, InterruptedException {
+ flushStore();
+ w.end(extraMetaData);
+ }
+
+ public void write(T value) throws IOException, InterruptedException {
+ writeSupport.write(value);
+ ++ recordCount;
+ checkBlockSizeReached();
+ }
+
+ private void checkBlockSizeReached() throws IOException {
+ if (recordCount >= recordCountForNextMemCheck) { // checking the memory size is relatively expensive, so let's not do it for every record.
+ long memSize = store.memSize();
+ if (memSize > blockSize) {
+ LOG.info(format("mem size %,d > %,d: flushing %,d records to disk.", memSize, blockSize, recordCount));
+ flushStore();
+ initStore();
+ recordCountForNextMemCheck = min(max(MINIMUM_RECORD_COUNT_FOR_CHECK, recordCount / 2), MAXIMUM_RECORD_COUNT_FOR_CHECK);
+ } else {
+ float recordSize = (float) memSize / recordCount;
+ recordCountForNextMemCheck = min(
+ max(MINIMUM_RECORD_COUNT_FOR_CHECK, (recordCount + (long)(blockSize / recordSize)) / 2), // will check halfway
+ recordCount + MAXIMUM_RECORD_COUNT_FOR_CHECK // will not look more than max records ahead
+ );
+ if (DEBUG) LOG.debug(format("Checked mem at %,d will check again at: %,d ", recordCount, recordCountForNextMemCheck));
+ }
+ }
+ }
+
+ private void flushStore()
+ throws IOException {
+ LOG.info(format("Flushing mem store to file. allocated memory: %,d", store.allocatedSize()));
+ if (store.allocatedSize() > 3 * blockSize) {
+ LOG.warn("Too much memory used: " + store.memUsageString());
+ }
+ w.startBlock(recordCount);
+ store.flush();
+ pageStore.flushToFileWriter(w);
+ recordCount = 0;
+ w.endBlock();
+ store = null;
+ pageStore = null;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetFileWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetFileWriter.java b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetFileWriter.java
new file mode 100644
index 0000000..73ce7c2
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetFileWriter.java
@@ -0,0 +1,504 @@
+/**
+ * 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.storage.thirdparty.parquet;
+
+import static parquet.Log.DEBUG;
+import static parquet.format.Util.writeFileMetaData;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import parquet.Log;
+import parquet.Version;
+import parquet.bytes.BytesInput;
+import parquet.bytes.BytesUtils;
+import parquet.column.ColumnDescriptor;
+import parquet.column.page.DictionaryPage;
+import parquet.column.statistics.Statistics;
+import parquet.format.converter.ParquetMetadataConverter;
+import parquet.hadoop.Footer;
+import parquet.hadoop.metadata.BlockMetaData;
+import parquet.hadoop.metadata.ColumnChunkMetaData;
+import parquet.hadoop.metadata.ColumnPath;
+import parquet.hadoop.metadata.CompressionCodecName;
+import parquet.hadoop.metadata.FileMetaData;
+import parquet.hadoop.metadata.GlobalMetaData;
+import parquet.hadoop.metadata.ParquetMetadata;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.MessageType;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * Internal implementation of the Parquet file writer as a block container
+ *
+ * @author Julien Le Dem
+ *
+ */
+public class ParquetFileWriter {
+ private static final Log LOG = Log.getLog(ParquetFileWriter.class);
+
+ public static final String PARQUET_METADATA_FILE = "_metadata";
+ public static final byte[] MAGIC = "PAR1".getBytes(Charset.forName("ASCII"));
+ public static final int CURRENT_VERSION = 1;
+
+ private static final ParquetMetadataConverter metadataConverter = new ParquetMetadataConverter();
+
+ private final MessageType schema;
+ private final FSDataOutputStream out;
+ private BlockMetaData currentBlock;
+ private ColumnChunkMetaData currentColumn;
+ private long currentRecordCount;
+ private List<BlockMetaData> blocks = new ArrayList<BlockMetaData>();
+ private long uncompressedLength;
+ private long compressedLength;
+ private Set<parquet.column.Encoding> currentEncodings;
+
+ private CompressionCodecName currentChunkCodec;
+ private ColumnPath currentChunkPath;
+ private PrimitiveTypeName currentChunkType;
+ private long currentChunkFirstDataPage;
+ private long currentChunkDictionaryPageOffset;
+ private long currentChunkValueCount;
+
+ private Statistics currentStatistics;
+
+ /**
+ * Captures the order in which methods should be called
+ *
+ * @author Julien Le Dem
+ *
+ */
+ private enum STATE {
+ NOT_STARTED {
+ STATE start() {
+ return STARTED;
+ }
+ },
+ STARTED {
+ STATE startBlock() {
+ return BLOCK;
+ }
+ STATE end() {
+ return ENDED;
+ }
+ },
+ BLOCK {
+ STATE startColumn() {
+ return COLUMN;
+ }
+ STATE endBlock() {
+ return STARTED;
+ }
+ },
+ COLUMN {
+ STATE endColumn() {
+ return BLOCK;
+ };
+ STATE write() {
+ return this;
+ }
+ },
+ ENDED;
+
+ STATE start() throws IOException { return error(); }
+ STATE startBlock() throws IOException { return error(); }
+ STATE startColumn() throws IOException { return error(); }
+ STATE write() throws IOException { return error(); }
+ STATE endColumn() throws IOException { return error(); }
+ STATE endBlock() throws IOException { return error(); }
+ STATE end() throws IOException { return error(); }
+
+ private final STATE error() throws IOException {
+ throw new IOException("The file being written is in an invalid state. Probably caused by an error thrown previously. Current state: " + this.name());
+ }
+ }
+
+ private STATE state = STATE.NOT_STARTED;
+
+ /**
+ *
+ * @param schema the schema of the data
+ * @param out the file to write to
+ * @param codec the codec to use to compress blocks
+ * @throws IOException if the file can not be created
+ */
+ public ParquetFileWriter(Configuration configuration, MessageType schema, Path file) throws IOException {
+ super();
+ this.schema = schema;
+ FileSystem fs = file.getFileSystem(configuration);
+ this.out = fs.create(file, false);
+ }
+
+ /**
+ * start the file
+ * @throws IOException
+ */
+ public void start() throws IOException {
+ state = state.start();
+ if (DEBUG) LOG.debug(out.getPos() + ": start");
+ out.write(MAGIC);
+ }
+
+ /**
+ * start a block
+ * @param recordCount the record count in this block
+ * @throws IOException
+ */
+ public void startBlock(long recordCount) throws IOException {
+ state = state.startBlock();
+ if (DEBUG) LOG.debug(out.getPos() + ": start block");
+// out.write(MAGIC); // TODO: add a magic delimiter
+ currentBlock = new BlockMetaData();
+ currentRecordCount = recordCount;
+ }
+
+ /**
+ * start a column inside a block
+ * @param descriptor the column descriptor
+ * @param valueCount the value count in this column
+ * @param statistics the statistics in this column
+ * @param compressionCodecName
+ * @throws IOException
+ */
+ public void startColumn(ColumnDescriptor descriptor,
+ long valueCount,
+ CompressionCodecName compressionCodecName) throws IOException {
+ state = state.startColumn();
+ if (DEBUG) LOG.debug(out.getPos() + ": start column: " + descriptor + " count=" + valueCount);
+ currentEncodings = new HashSet<parquet.column.Encoding>();
+ currentChunkPath = ColumnPath.get(descriptor.getPath());
+ currentChunkType = descriptor.getType();
+ currentChunkCodec = compressionCodecName;
+ currentChunkValueCount = valueCount;
+ currentChunkFirstDataPage = out.getPos();
+ compressedLength = 0;
+ uncompressedLength = 0;
+ // need to know what type of stats to initialize to
+ // better way to do this?
+ currentStatistics = Statistics.getStatsBasedOnType(currentChunkType);
+ }
+
+ /**
+ * writes a dictionary page page
+ * @param dictionaryPage the dictionary page
+ */
+ public void writeDictionaryPage(DictionaryPage dictionaryPage) throws IOException {
+ state = state.write();
+ if (DEBUG) LOG.debug(out.getPos() + ": write dictionary page: " + dictionaryPage.getDictionarySize() + " values");
+ currentChunkDictionaryPageOffset = out.getPos();
+ int uncompressedSize = dictionaryPage.getUncompressedSize();
+ int compressedPageSize = (int)dictionaryPage.getBytes().size(); // TODO: fix casts
+ metadataConverter.writeDictionaryPageHeader(
+ uncompressedSize,
+ compressedPageSize,
+ dictionaryPage.getDictionarySize(),
+ dictionaryPage.getEncoding(),
+ out);
+ long headerSize = out.getPos() - currentChunkDictionaryPageOffset;
+ this.uncompressedLength += uncompressedSize + headerSize;
+ this.compressedLength += compressedPageSize + headerSize;
+ if (DEBUG) LOG.debug(out.getPos() + ": write dictionary page content " + compressedPageSize);
+ dictionaryPage.getBytes().writeAllTo(out);
+ currentEncodings.add(dictionaryPage.getEncoding());
+ }
+
+
+ /**
+ * writes a single page
+ * @param valueCount count of values
+ * @param uncompressedPageSize the size of the data once uncompressed
+ * @param bytes the compressed data for the page without header
+ * @param rlEncoding encoding of the repetition level
+ * @param dlEncoding encoding of the definition level
+ * @param valuesEncoding encoding of values
+ */
+ @Deprecated
+ public void writeDataPage(
+ int valueCount, int uncompressedPageSize,
+ BytesInput bytes,
+ parquet.column.Encoding rlEncoding,
+ parquet.column.Encoding dlEncoding,
+ parquet.column.Encoding valuesEncoding) throws IOException {
+ state = state.write();
+ long beforeHeader = out.getPos();
+ if (DEBUG) LOG.debug(beforeHeader + ": write data page: " + valueCount + " values");
+ int compressedPageSize = (int)bytes.size();
+ metadataConverter.writeDataPageHeader(
+ uncompressedPageSize, compressedPageSize,
+ valueCount,
+ rlEncoding,
+ dlEncoding,
+ valuesEncoding,
+ out);
+ long headerSize = out.getPos() - beforeHeader;
+ this.uncompressedLength += uncompressedPageSize + headerSize;
+ this.compressedLength += compressedPageSize + headerSize;
+ if (DEBUG) LOG.debug(out.getPos() + ": write data page content " + compressedPageSize);
+ bytes.writeAllTo(out);
+ currentEncodings.add(rlEncoding);
+ currentEncodings.add(dlEncoding);
+ currentEncodings.add(valuesEncoding);
+ }
+
+ /**
+ * writes a single page
+ * @param valueCount count of values
+ * @param uncompressedPageSize the size of the data once uncompressed
+ * @param bytes the compressed data for the page without header
+ * @param rlEncoding encoding of the repetition level
+ * @param dlEncoding encoding of the definition level
+ * @param valuesEncoding encoding of values
+ */
+ public void writeDataPage(
+ int valueCount, int uncompressedPageSize,
+ BytesInput bytes,
+ Statistics statistics,
+ parquet.column.Encoding rlEncoding,
+ parquet.column.Encoding dlEncoding,
+ parquet.column.Encoding valuesEncoding) throws IOException {
+ state = state.write();
+ long beforeHeader = out.getPos();
+ if (DEBUG) LOG.debug(beforeHeader + ": write data page: " + valueCount + " values");
+ int compressedPageSize = (int)bytes.size();
+ metadataConverter.writeDataPageHeader(
+ uncompressedPageSize, compressedPageSize,
+ valueCount,
+ statistics,
+ rlEncoding,
+ dlEncoding,
+ valuesEncoding,
+ out);
+ long headerSize = out.getPos() - beforeHeader;
+ this.uncompressedLength += uncompressedPageSize + headerSize;
+ this.compressedLength += compressedPageSize + headerSize;
+ if (DEBUG) LOG.debug(out.getPos() + ": write data page content " + compressedPageSize);
+ bytes.writeAllTo(out);
+ currentStatistics.mergeStatistics(statistics);
+ currentEncodings.add(rlEncoding);
+ currentEncodings.add(dlEncoding);
+ currentEncodings.add(valuesEncoding);
+ }
+
+ /**
+ * writes a number of pages at once
+ * @param bytes bytes to be written including page headers
+ * @param uncompressedTotalPageSize total uncompressed size (without page headers)
+ * @param compressedTotalPageSize total compressed size (without page headers)
+ * @throws IOException
+ */
+ void writeDataPages(BytesInput bytes,
+ long uncompressedTotalPageSize,
+ long compressedTotalPageSize,
+ Statistics totalStats,
+ List<parquet.column.Encoding> encodings) throws IOException {
+ state = state.write();
+ if (DEBUG) LOG.debug(out.getPos() + ": write data pages");
+ long headersSize = bytes.size() - compressedTotalPageSize;
+ this.uncompressedLength += uncompressedTotalPageSize + headersSize;
+ this.compressedLength += compressedTotalPageSize + headersSize;
+ if (DEBUG) LOG.debug(out.getPos() + ": write data pages content");
+ bytes.writeAllTo(out);
+ currentEncodings.addAll(encodings);
+ currentStatistics = totalStats;
+ }
+
+ /**
+ * end a column (once all rep, def and data have been written)
+ * @throws IOException
+ */
+ public void endColumn() throws IOException {
+ state = state.endColumn();
+ if (DEBUG) LOG.debug(out.getPos() + ": end column");
+ currentBlock.addColumn(ColumnChunkMetaData.get(
+ currentChunkPath,
+ currentChunkType,
+ currentChunkCodec,
+ currentEncodings,
+ currentStatistics,
+ currentChunkFirstDataPage,
+ currentChunkDictionaryPageOffset,
+ currentChunkValueCount,
+ compressedLength,
+ uncompressedLength));
+ if (DEBUG) LOG.info("ended Column chumk: " + currentColumn);
+ currentColumn = null;
+ this.currentBlock.setTotalByteSize(currentBlock.getTotalByteSize() + uncompressedLength);
+ this.uncompressedLength = 0;
+ this.compressedLength = 0;
+ }
+
+ /**
+ * ends a block once all column chunks have been written
+ * @throws IOException
+ */
+ public void endBlock() throws IOException {
+ state = state.endBlock();
+ if (DEBUG) LOG.debug(out.getPos() + ": end block");
+ currentBlock.setRowCount(currentRecordCount);
+ blocks.add(currentBlock);
+ currentBlock = null;
+ }
+
+ /**
+ * ends a file once all blocks have been written.
+ * closes the file.
+ * @param extraMetaData the extra meta data to write in the footer
+ * @throws IOException
+ */
+ public void end(Map<String, String> extraMetaData) throws IOException {
+ state = state.end();
+ if (DEBUG) LOG.debug(out.getPos() + ": end");
+ ParquetMetadata footer = new ParquetMetadata(new FileMetaData(schema, extraMetaData, Version.FULL_VERSION), blocks);
+ serializeFooter(footer, out);
+ out.close();
+ }
+
+ private static void serializeFooter(ParquetMetadata footer, FSDataOutputStream out) throws IOException {
+ long footerIndex = out.getPos();
+ parquet.format.FileMetaData parquetMetadata = new ParquetMetadataConverter().toParquetMetadata(CURRENT_VERSION, footer);
+ writeFileMetaData(parquetMetadata, out);
+ if (DEBUG) LOG.debug(out.getPos() + ": footer length = " + (out.getPos() - footerIndex));
+ BytesUtils.writeIntLittleEndian(out, (int)(out.getPos() - footerIndex));
+ out.write(MAGIC);
+ }
+
+ /**
+ * writes a _metadata file
+ * @param configuration the configuration to use to get the FileSystem
+ * @param outputPath the directory to write the _metadata file to
+ * @param footers the list of footers to merge
+ * @throws IOException
+ */
+ public static void writeMetadataFile(Configuration configuration, Path outputPath, List<Footer> footers) throws IOException {
+ Path metaDataPath = new Path(outputPath, PARQUET_METADATA_FILE);
+ FileSystem fs = outputPath.getFileSystem(configuration);
+ outputPath = outputPath.makeQualified(fs);
+ FSDataOutputStream metadata = fs.create(metaDataPath);
+ metadata.write(MAGIC);
+ ParquetMetadata metadataFooter = mergeFooters(outputPath, footers);
+ serializeFooter(metadataFooter, metadata);
+ metadata.close();
+ }
+
+ private static ParquetMetadata mergeFooters(Path root, List<Footer> footers) {
+ String rootPath = root.toString();
+ GlobalMetaData fileMetaData = null;
+ List<BlockMetaData> blocks = new ArrayList<BlockMetaData>();
+ for (Footer footer : footers) {
+ String path = footer.getFile().toString();
+ if (!path.startsWith(rootPath)) {
+ throw new ParquetEncodingException(path + " invalid: all the files must be contained in the root " + root);
+ }
+ path = path.substring(rootPath.length());
+ while (path.startsWith("/")) {
+ path = path.substring(1);
+ }
+ fileMetaData = mergeInto(footer.getParquetMetadata().getFileMetaData(), fileMetaData);
+ for (BlockMetaData block : footer.getParquetMetadata().getBlocks()) {
+ block.setPath(path);
+ blocks.add(block);
+ }
+ }
+ return new ParquetMetadata(fileMetaData.merge(), blocks);
+ }
+
+ /**
+ * @return the current position in the underlying file
+ * @throws IOException
+ */
+ public long getPos() throws IOException {
+ return out.getPos();
+ }
+
+ /**
+ * Will merge the metadata of all the footers together
+ * @param footers the list files footers to merge
+ * @return the global meta data for all the footers
+ */
+ static GlobalMetaData getGlobalMetaData(List<Footer> footers) {
+ GlobalMetaData fileMetaData = null;
+ for (Footer footer : footers) {
+ ParquetMetadata currentMetadata = footer.getParquetMetadata();
+ fileMetaData = mergeInto(currentMetadata.getFileMetaData(), fileMetaData);
+ }
+ return fileMetaData;
+ }
+
+ /**
+ * Will return the result of merging toMerge into mergedMetadata
+ * @param toMerge the metadata toMerge
+ * @param mergedMetadata the reference metadata to merge into
+ * @return the result of the merge
+ */
+ static GlobalMetaData mergeInto(
+ FileMetaData toMerge,
+ GlobalMetaData mergedMetadata) {
+ MessageType schema = null;
+ Map<String, Set<String>> newKeyValues = new HashMap<String, Set<String>>();
+ Set<String> createdBy = new HashSet<String>();
+ if (mergedMetadata != null) {
+ schema = mergedMetadata.getSchema();
+ newKeyValues.putAll(mergedMetadata.getKeyValueMetaData());
+ createdBy.addAll(mergedMetadata.getCreatedBy());
+ }
+ if ((schema == null && toMerge.getSchema() != null)
+ || (schema != null && !schema.equals(toMerge.getSchema()))) {
+ schema = mergeInto(toMerge.getSchema(), schema);
+ }
+ for (Entry<String, String> entry : toMerge.getKeyValueMetaData().entrySet()) {
+ Set<String> values = newKeyValues.get(entry.getKey());
+ if (values == null) {
+ values = new HashSet<String>();
+ newKeyValues.put(entry.getKey(), values);
+ }
+ values.add(entry.getValue());
+ }
+ createdBy.add(toMerge.getCreatedBy());
+ return new GlobalMetaData(
+ schema,
+ newKeyValues,
+ createdBy);
+ }
+
+ /**
+ * will return the result of merging toMerge into mergedSchema
+ * @param toMerge the schema to merge into mergedSchema
+ * @param mergedSchema the schema to append the fields to
+ * @return the resulting schema
+ */
+ static MessageType mergeInto(MessageType toMerge, MessageType mergedSchema) {
+ if (mergedSchema == null) {
+ return toMerge;
+ }
+ return mergedSchema.union(toMerge);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetReader.java b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetReader.java
new file mode 100644
index 0000000..0fb2c3a
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetReader.java
@@ -0,0 +1,151 @@
+/**
+ * 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.storage.thirdparty.parquet;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import parquet.filter.UnboundRecordFilter;
+import parquet.hadoop.*;
+import parquet.hadoop.api.InitContext;
+import parquet.hadoop.api.ReadSupport;
+import parquet.hadoop.api.ReadSupport.ReadContext;
+import parquet.hadoop.metadata.BlockMetaData;
+import parquet.hadoop.metadata.GlobalMetaData;
+import parquet.schema.MessageType;
+
+/**
+ * Read records from a Parquet file.
+ */
+public class ParquetReader<T> implements Closeable {
+
+ private ReadSupport<T> readSupport;
+ private UnboundRecordFilter filter;
+ private Configuration conf;
+ private ReadContext readContext;
+ private Iterator<Footer> footersIterator;
+ private InternalParquetRecordReader<T> reader;
+ private GlobalMetaData globalMetaData;
+
+ /**
+ * @param file the file to read
+ * @param readSupport to materialize records
+ * @throws IOException
+ */
+ public ParquetReader(Path file, ReadSupport<T> readSupport) throws IOException {
+ this(file, readSupport, null);
+ }
+
+ /**
+ * @param conf the configuration
+ * @param file the file to read
+ * @param readSupport to materialize records
+ * @throws IOException
+ */
+ public ParquetReader(Configuration conf, Path file, ReadSupport<T> readSupport) throws IOException {
+ this(conf, file, readSupport, null);
+ }
+
+ /**
+ * @param file the file to read
+ * @param readSupport to materialize records
+ * @param filter the filter to use to filter records
+ * @throws IOException
+ */
+ public ParquetReader(Path file, ReadSupport<T> readSupport, UnboundRecordFilter filter) throws IOException {
+ this(new Configuration(), file, readSupport, filter);
+ }
+
+ /**
+ * @param conf the configuration
+ * @param file the file to read
+ * @param readSupport to materialize records
+ * @param filter the filter to use to filter records
+ * @throws IOException
+ */
+ public ParquetReader(Configuration conf, Path file, ReadSupport<T> readSupport, UnboundRecordFilter filter) throws IOException {
+ this.readSupport = readSupport;
+ this.filter = filter;
+ this.conf = conf;
+
+ FileSystem fs = file.getFileSystem(conf);
+ List<FileStatus> statuses = Arrays.asList(fs.listStatus(file));
+ List<Footer> footers = ParquetFileReader.readAllFootersInParallelUsingSummaryFiles(conf, statuses);
+ this.footersIterator = footers.iterator();
+ globalMetaData = ParquetFileWriter.getGlobalMetaData(footers);
+
+ List<BlockMetaData> blocks = new ArrayList<BlockMetaData>();
+ for (Footer footer : footers) {
+ blocks.addAll(footer.getParquetMetadata().getBlocks());
+ }
+
+ MessageType schema = globalMetaData.getSchema();
+ Map<String, Set<String>> extraMetadata = globalMetaData.getKeyValueMetaData();
+ readContext = readSupport.init(new InitContext(conf, extraMetadata, schema));
+ }
+
+ /**
+ * @return the next record or null if finished
+ * @throws IOException
+ */
+ public T read() throws IOException {
+ try {
+ if (reader != null && reader.nextKeyValue()) {
+ return reader.getCurrentValue();
+ } else {
+ initReader();
+ return reader == null ? null : read();
+ }
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+ private void initReader() throws IOException {
+ if (reader != null) {
+ reader.close();
+ reader = null;
+ }
+ if (footersIterator.hasNext()) {
+ Footer footer = footersIterator.next();
+ reader = new InternalParquetRecordReader<T>(readSupport, filter);
+ reader.initialize(
+ readContext.getRequestedSchema(), globalMetaData.getSchema(), footer.getParquetMetadata().getFileMetaData().getKeyValueMetaData(),
+ readContext.getReadSupportMetadata(), footer.getFile(), footer.getParquetMetadata().getBlocks(), conf);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (reader != null) {
+ reader.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/9cf10719/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetWriter.java b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetWriter.java
new file mode 100644
index 0000000..743d168
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/thirdparty/parquet/ParquetWriter.java
@@ -0,0 +1,220 @@
+/**
+ * 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.storage.thirdparty.parquet;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import parquet.column.ParquetProperties;
+import parquet.hadoop.api.WriteSupport;
+import parquet.hadoop.metadata.CompressionCodecName;
+import parquet.schema.MessageType;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+public class ParquetWriter<T> implements Closeable {
+
+ public static final int DEFAULT_BLOCK_SIZE = 128 * 1024 * 1024;
+ public static final int DEFAULT_PAGE_SIZE = 1 * 1024 * 1024;
+ public static final CompressionCodecName DEFAULT_COMPRESSION_CODEC_NAME =
+ CompressionCodecName.UNCOMPRESSED;
+ public static final boolean DEFAULT_IS_DICTIONARY_ENABLED = true;
+ public static final boolean DEFAULT_IS_VALIDATING_ENABLED = false;
+ public static final ParquetProperties.WriterVersion DEFAULT_WRITER_VERSION =
+ ParquetProperties.WriterVersion.PARQUET_1_0;
+
+ private final InternalParquetRecordWriter<T> writer;
+
+ /**
+ * Create a new ParquetWriter.
+ * (with dictionary encoding enabled and validation off)
+ *
+ * @param file the file to create
+ * @param writeSupport the implementation to write a record to a RecordConsumer
+ * @param compressionCodecName the compression codec to use
+ * @param blockSize the block size threshold
+ * @param pageSize the page size threshold
+ * @throws java.io.IOException
+ * @see #ParquetWriter(org.apache.hadoop.fs.Path, parquet.hadoop.api.WriteSupport, CompressionCodecName, int, int, boolean, boolean)
+ */
+ public ParquetWriter(Path file, WriteSupport<T> writeSupport, CompressionCodecName compressionCodecName, int blockSize, int pageSize) throws IOException {
+ this(file, writeSupport, compressionCodecName, blockSize, pageSize,
+ DEFAULT_IS_DICTIONARY_ENABLED, DEFAULT_IS_VALIDATING_ENABLED);
+ }
+
+ /**
+ * Create a new ParquetWriter.
+ *
+ * @param file the file to create
+ * @param writeSupport the implementation to write a record to a RecordConsumer
+ * @param compressionCodecName the compression codec to use
+ * @param blockSize the block size threshold
+ * @param pageSize the page size threshold (both data and dictionary)
+ * @param enableDictionary to turn dictionary encoding on
+ * @param validating to turn on validation using the schema
+ * @throws IOException
+ * @see #ParquetWriter(Path, WriteSupport, CompressionCodecName, int, int, int, boolean, boolean)
+ */
+ public ParquetWriter(
+ Path file,
+ WriteSupport<T> writeSupport,
+ CompressionCodecName compressionCodecName,
+ int blockSize,
+ int pageSize,
+ boolean enableDictionary,
+ boolean validating) throws IOException {
+ this(file, writeSupport, compressionCodecName, blockSize, pageSize, pageSize, enableDictionary, validating);
+ }
+
+ /**
+ * Create a new ParquetWriter.
+ *
+ * @param file the file to create
+ * @param writeSupport the implementation to write a record to a RecordConsumer
+ * @param compressionCodecName the compression codec to use
+ * @param blockSize the block size threshold
+ * @param pageSize the page size threshold
+ * @param dictionaryPageSize the page size threshold for the dictionary pages
+ * @param enableDictionary to turn dictionary encoding on
+ * @param validating to turn on validation using the schema
+ * @throws IOException
+ * @see #ParquetWriter(Path, WriteSupport, CompressionCodecName, int, int, int, boolean, boolean, parquet.column.ParquetProperties.WriterVersion)
+ */
+ public ParquetWriter(
+ Path file,
+ WriteSupport<T> writeSupport,
+ CompressionCodecName compressionCodecName,
+ int blockSize,
+ int pageSize,
+ int dictionaryPageSize,
+ boolean enableDictionary,
+ boolean validating) throws IOException {
+ this(file, writeSupport, compressionCodecName, blockSize, pageSize,
+ dictionaryPageSize, enableDictionary, validating,
+ DEFAULT_WRITER_VERSION);
+ }
+
+ /**
+ * Create a new ParquetWriter.
+ *
+ * Directly instantiates a Hadoop {@link org.apache.hadoop.conf.Configuration} which reads
+ * configuration from the classpath.
+ *
+ * @param file the file to create
+ * @param writeSupport the implementation to write a record to a RecordConsumer
+ * @param compressionCodecName the compression codec to use
+ * @param blockSize the block size threshold
+ * @param pageSize the page size threshold
+ * @param dictionaryPageSize the page size threshold for the dictionary pages
+ * @param enableDictionary to turn dictionary encoding on
+ * @param validating to turn on validation using the schema
+ * @param writerVersion version of parquetWriter from {@link ParquetProperties.WriterVersion}
+ * @throws IOException
+ * @see #ParquetWriter(Path, WriteSupport, CompressionCodecName, int, int, int, boolean, boolean, parquet.column.ParquetProperties.WriterVersion, org.apache.hadoop.conf.Configuration)
+ */
+ public ParquetWriter(
+ Path file,
+ WriteSupport<T> writeSupport,
+ CompressionCodecName compressionCodecName,
+ int blockSize,
+ int pageSize,
+ int dictionaryPageSize,
+ boolean enableDictionary,
+ boolean validating,
+ ParquetProperties.WriterVersion writerVersion) throws IOException {
+ this(file, writeSupport, compressionCodecName, blockSize, pageSize, dictionaryPageSize, enableDictionary, validating, writerVersion, new Configuration());
+ }
+
+ /**
+ * Create a new ParquetWriter.
+ *
+ * @param file the file to create
+ * @param writeSupport the implementation to write a record to a RecordConsumer
+ * @param compressionCodecName the compression codec to use
+ * @param blockSize the block size threshold
+ * @param pageSize the page size threshold
+ * @param dictionaryPageSize the page size threshold for the dictionary pages
+ * @param enableDictionary to turn dictionary encoding on
+ * @param validating to turn on validation using the schema
+ * @param writerVersion version of parquetWriter from {@link ParquetProperties.WriterVersion}
+ * @param conf Hadoop configuration to use while accessing the filesystem
+ * @throws IOException
+ */
+ public ParquetWriter(
+ Path file,
+ WriteSupport<T> writeSupport,
+ CompressionCodecName compressionCodecName,
+ int blockSize,
+ int pageSize,
+ int dictionaryPageSize,
+ boolean enableDictionary,
+ boolean validating,
+ ParquetProperties.WriterVersion writerVersion,
+ Configuration conf) throws IOException {
+
+ WriteSupport.WriteContext writeContext = writeSupport.init(conf);
+ MessageType schema = writeContext.getSchema();
+
+ ParquetFileWriter fileWriter = new ParquetFileWriter(conf, schema, file);
+ fileWriter.start();
+
+ CodecFactory codecFactory = new CodecFactory(conf);
+ CodecFactory.BytesCompressor compressor = codecFactory.getCompressor(compressionCodecName, 0);
+ this.writer = new InternalParquetRecordWriter<T>(
+ fileWriter,
+ writeSupport,
+ schema,
+ writeContext.getExtraMetaData(),
+ blockSize,
+ pageSize,
+ compressor,
+ dictionaryPageSize,
+ enableDictionary,
+ validating,
+ writerVersion);
+ }
+
+ /**
+ * Create a new ParquetWriter. The default block size is 50 MB.The default
+ * page size is 1 MB. Default compression is no compression. Dictionary encoding is disabled.
+ *
+ * @param file the file to create
+ * @param writeSupport the implementation to write a record to a RecordConsumer
+ * @throws IOException
+ */
+ public ParquetWriter(Path file, WriteSupport<T> writeSupport) throws IOException {
+ this(file, writeSupport, DEFAULT_COMPRESSION_CODEC_NAME, DEFAULT_BLOCK_SIZE, DEFAULT_PAGE_SIZE);
+ }
+
+ public void write(T object) throws IOException {
+ try {
+ writer.write(object);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ writer.close();
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }}
[15/19] git commit: TAJO-673: Assign proper number of tasks when
inserting into partitioned table. (jaehwa)
Posted by ji...@apache.org.
TAJO-673: Assign proper number of tasks when inserting into partitioned table. (jaehwa)
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/48dbfd92
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/48dbfd92
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/48dbfd92
Branch: refs/heads/index_support
Commit: 48dbfd92c4f424ffef6c9aa03e905882bd42bbec
Parents: 5e2d6c3
Author: blrunner <jh...@gruter.com>
Authored: Tue Jul 15 12:35:00 2014 +0900
Committer: blrunner <jh...@gruter.com>
Committed: Tue Jul 15 12:35:00 2014 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../java/org/apache/tajo/conf/TajoConf.java | 3 +
.../engine/planner/PhysicalPlannerImpl.java | 1 +
.../engine/planner/global/GlobalPlanner.java | 3 +-
.../physical/HashShuffleFileWriteExec.java | 1 +
.../tajo/master/querymaster/QueryUnit.java | 13 +++
.../master/querymaster/QueryUnitAttempt.java | 2 +-
.../tajo/master/querymaster/Repartitioner.java | 113 ++++++++++++++++++-
.../main/java/org/apache/tajo/worker/Task.java | 11 ++
.../apache/tajo/worker/TaskAttemptContext.java | 18 +++
.../src/main/proto/TajoWorkerProtocol.proto | 2 +
.../tajo/engine/query/TestTablePartitions.java | 108 ++++++++++++++++--
.../tajo/pullserver/PullServerAuxService.java | 4 +-
.../tajo/pullserver/TajoPullServerService.java | 4 +-
14 files changed, 264 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index c0dcbe3..3803ded 100644
--- a/CHANGES
+++ b/CHANGES
@@ -84,6 +84,8 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-673: Assign proper number of tasks when inserting into partitioned table. (jaehwa)
+
TAJO-916: SubQuery::computeStatFromTasks occasionally fail.
(Hyoungjun Kim via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index dd5327d..83ff9ed 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -249,6 +249,9 @@ public class TajoConf extends Configuration {
DIST_QUERY_SORT_PARTITION_VOLUME("tajo.dist-query.sort.partition-volume-mb", 256),
DIST_QUERY_GROUPBY_PARTITION_VOLUME("tajo.dist-query.groupby.partition-volume-mb", 256),
+ DIST_QUERY_TABLE_PARTITION_VOLUME("tajo.dist-query.table-partition.task-volume-mb",
+ 256 * 1024 * 1024),
+
//////////////////////////////////
// Physical Executors
//////////////////////////////////
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index cf02ecd..6678e46 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -732,6 +732,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
ShuffleFileWriteNode plan, PhysicalExec subOp) throws IOException {
switch (plan.getShuffleType()) {
case HASH_SHUFFLE:
+ case SCATTERED_HASH_SHUFFLE:
return new HashShuffleFileWriteExec(ctx, sm, plan, subOp);
case RANGE_SHUFFLE:
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
index 4e27574..69ecd02 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
@@ -1149,10 +1149,11 @@ public class GlobalPlanner {
shuffleKeys[i++] = insertNode.getProjectedSchema().getColumn(id);
}
channel.setShuffleKeys(shuffleKeys);
+ channel.setShuffleType(SCATTERED_HASH_SHUFFLE);
} else {
channel.setShuffleKeys(partitionMethod.getExpressionSchema().toArray());
+ channel.setShuffleType(HASH_SHUFFLE);
}
- channel.setShuffleType(HASH_SHUFFLE);
channel.setShuffleOutputNum(32);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
index 678b745..44e8646 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
@@ -129,6 +129,7 @@ public final class HashShuffleFileWriteExec extends UnaryPhysicalExec {
statSet.add(app.getStats());
if (app.getStats().getNumRows() > 0) {
context.addShuffleFileOutput(partNum, getDataFile(partNum).getName());
+ context.addPartitionOutputVolume(partNum, app.getStats().getNumBytes());
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
index 6cada07..806c0f1 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
@@ -667,6 +667,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
int attemptId;
int partId;
PullHost host;
+ long volume;
public IntermediateEntry(int taskId, int attemptId, int partId, PullHost host) {
this.taskId = taskId;
@@ -675,6 +676,14 @@ public class QueryUnit implements EventHandler<TaskEvent> {
this.host = host;
}
+ public IntermediateEntry(int taskId, int attemptId, int partId, PullHost host, long volume) {
+ this.taskId = taskId;
+ this.attemptId = attemptId;
+ this.partId = partId;
+ this.host = host;
+ this.volume = volume;
+ }
+
public ExecutionBlockId getEbId() {
return ebId;
}
@@ -699,6 +708,10 @@ public class QueryUnit implements EventHandler<TaskEvent> {
return this.host;
}
+ public long getVolume() {
+ return this.volume;
+ }
+
@Override
public int hashCode() {
return Objects.hashCode(ebId, taskId, partId, attemptId, host);
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
index 361f88f..a4fa12f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnitAttempt.java
@@ -280,7 +280,7 @@ public class QueryUnitAttempt implements EventHandler<TaskAttemptEvent> {
PullHost host = new PullHost(getHost(), getPullServerPort());
for (ShuffleFileOutput p : report.getShuffleFileOutputsList()) {
IntermediateEntry entry = new IntermediateEntry(getId().getQueryUnitId().getId(),
- getId().getId(), p.getPartId(), host);
+ getId().getId(), p.getPartId(), host, p.getVolume());
partitions.add(entry);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
index ce2194e..1cc5b78 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
@@ -19,6 +19,7 @@
package org.apache.tajo.master.querymaster;
import com.google.common.collect.Lists;
+import com.google.common.primitives.Ints;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
@@ -60,6 +61,7 @@ import java.util.Map.Entry;
import static org.apache.tajo.ipc.TajoWorkerProtocol.ShuffleType.HASH_SHUFFLE;
import static org.apache.tajo.ipc.TajoWorkerProtocol.ShuffleType.RANGE_SHUFFLE;
+import static org.apache.tajo.ipc.TajoWorkerProtocol.ShuffleType.SCATTERED_HASH_SHUFFLE;
/**
* Repartitioner creates non-leaf tasks and shuffles intermediate data.
@@ -529,7 +531,8 @@ public class Repartitioner {
MasterPlan masterPlan, SubQuery subQuery, int maxNum)
throws IOException {
DataChannel channel = masterPlan.getIncomingChannels(subQuery.getBlock().getId()).get(0);
- if (channel.getShuffleType() == HASH_SHUFFLE) {
+ if (channel.getShuffleType() == HASH_SHUFFLE
+ || channel.getShuffleType() == SCATTERED_HASH_SHUFFLE) {
scheduleHashShuffledFetches(schedulerContext, masterPlan, subQuery, channel, maxNum);
} else if (channel.getShuffleType() == RANGE_SHUFFLE) {
scheduleRangeShuffledFetches(schedulerContext, masterPlan, subQuery, channel, maxNum);
@@ -668,12 +671,23 @@ public class Repartitioner {
Map<QueryUnit.PullHost, List<IntermediateEntry>> hashedByHost;
Map<Integer, Collection<FetchImpl>> finalFetches = new HashMap<Integer, Collection<FetchImpl>>();
+ Map<ExecutionBlockId, List<IntermediateEntry>> intermediates = new HashMap<ExecutionBlockId,
+ List<IntermediateEntry>>();
for (ExecutionBlock block : masterPlan.getChilds(execBlock)) {
List<IntermediateEntry> partitions = new ArrayList<IntermediateEntry>();
for (QueryUnit tasks : subQuery.getContext().getSubQuery(block.getId()).getQueryUnits()) {
if (tasks.getIntermediateData() != null) {
partitions.addAll(tasks.getIntermediateData());
+
+ // In scattered hash shuffle, Collecting each IntermediateEntry
+ if (channel.getShuffleType() == SCATTERED_HASH_SHUFFLE) {
+ if (intermediates.containsKey(block.getId())) {
+ intermediates.get(block.getId()).addAll(tasks.getIntermediateData());
+ } else {
+ intermediates.put(block.getId(), tasks.getIntermediateData());
+ }
+ }
}
}
Map<Integer, List<IntermediateEntry>> hashed = hashByKey(partitions);
@@ -709,12 +723,99 @@ public class Repartitioner {
}
// set the proper number of tasks to the estimated task num
- schedulerContext.setEstimatedTaskNum(determinedTaskNum);
- // divide fetch uris into the the proper number of tasks in a round robin manner.
- scheduleFetchesByRoundRobin(subQuery, finalFetches, scan.getTableName(), determinedTaskNum);
- LOG.info(subQuery.getId() + ", DeterminedTaskNum : " + determinedTaskNum);
+ if (channel.getShuffleType() == SCATTERED_HASH_SHUFFLE) {
+ scheduleScatteredHashShuffleFetches(schedulerContext, subQuery, intermediates,
+ scan.getTableName());
+ } else {
+ schedulerContext.setEstimatedTaskNum(determinedTaskNum);
+ // divide fetch uris into the the proper number of tasks in a round robin manner.
+ scheduleFetchesByRoundRobin(subQuery, finalFetches, scan.getTableName(), determinedTaskNum);
+ LOG.info(subQuery.getId() + ", DeterminedTaskNum : " + determinedTaskNum);
+ }
}
+ // Scattered hash shuffle hashes the key columns and groups the hash keys associated with
+ // the same hash key. Then, if the volume of a group is larger
+ // than DIST_QUERY_TABLE_PARTITION_VOLUME, it divides the group into more than two sub groups
+ // according to DIST_QUERY_TABLE_PARTITION_VOLUME (default size = 256MB).
+ // As a result, each group size always becomes the less than or equal
+ // to DIST_QUERY_TABLE_PARTITION_VOLUME. Finally, each subgroup is assigned to a query unit.
+ // It is usually used for writing partitioned tables.
+ public static void scheduleScatteredHashShuffleFetches(TaskSchedulerContext schedulerContext,
+ SubQuery subQuery, Map<ExecutionBlockId, List<IntermediateEntry>> intermediates,
+ String tableName) {
+ int i = 0;
+ int splitVolume = subQuery.getContext().getConf().
+ getIntVar(ConfVars.DIST_QUERY_TABLE_PARTITION_VOLUME);
+
+ long sumNumBytes = 0L;
+ Map<Integer, List<FetchImpl>> fetches = new HashMap<Integer, List<FetchImpl>>();
+
+ // Step 1 : divide fetch uris into the the proper number of tasks by
+ // SCATTERED_HASH_SHUFFLE_SPLIT_VOLUME
+ for (Entry<ExecutionBlockId, List<IntermediateEntry>> listEntry : intermediates.entrySet()) {
+
+ // Step 2: Sort IntermediateEntry by partition id. After first sort,
+ // we need to sort again by PullHost address because of data locality.
+ Collections.sort(listEntry.getValue(), new IntermediateEntryComparator());
+ for (IntermediateEntry interm : listEntry.getValue()) {
+ FetchImpl fetch = new FetchImpl(interm.getPullHost(), SCATTERED_HASH_SHUFFLE,
+ listEntry.getKey(), interm.getPartId(), TUtil.newList(interm));
+ if (fetches.size() == 0) {
+ fetches.put(i, TUtil.newList(fetch));
+ } else {
+
+ // Step 3: Compare current partition id with previous partition id because One task just
+ // can include one partitionId.
+ if (fetches.get(i).get(0).getPartitionId() != interm.getPartId()) {
+ i++;
+ fetches.put(i, TUtil.newList(fetch));
+ sumNumBytes = 0L;
+ } else {
+ if ((sumNumBytes + interm.getVolume()) < splitVolume) {
+ fetches.get(i).add(fetch);
+ } else {
+ i++;
+ fetches.put(i, TUtil.newList(fetch));
+ sumNumBytes = 0L;
+ }
+ }
+ }
+ sumNumBytes += interm.getVolume();
+ }
+ }
+
+ // Step 4 : Set the proper number of tasks to the estimated task num
+ schedulerContext.setEstimatedTaskNum(fetches.size());
+
+ // Step 5 : Apply divided fetches
+ i = 0;
+ Map<String, List<FetchImpl>>[] fetchesArray = new Map[fetches.size()];
+ for(Entry<Integer, List<FetchImpl>> entry : fetches.entrySet()) {
+ fetchesArray[i] = new HashMap<String, List<FetchImpl>>();
+ fetchesArray[i].put(tableName, entry.getValue());
+
+ SubQuery.scheduleFetches(subQuery, fetchesArray[i]);
+ i++;
+ }
+
+ LOG.info(subQuery.getId()
+ + ", ShuffleType:" + SCATTERED_HASH_SHUFFLE.name()
+ + ", DeterminedTaskNum : " + fetches.size());
+ }
+
+ static class IntermediateEntryComparator implements Comparator<IntermediateEntry> {
+
+ @Override
+ public int compare(IntermediateEntry o1, IntermediateEntry o2) {
+ int cmp = Ints.compare(o1.getPartId(), o2.getPartId());
+ if (cmp != 0) {
+ return cmp;
+ }
+
+ return o1.getPullHost().getHost().compareTo(o2.getPullHost().getHost());
+ }
+ }
public static List<URI> createFetchURL(FetchImpl fetch, boolean includeParts) {
String scheme = "http://";
@@ -729,6 +830,8 @@ public class Repartitioner {
urlPrefix.append("h");
} else if (fetch.getType() == RANGE_SHUFFLE) {
urlPrefix.append("r").append("&").append(fetch.getRangeParams());
+ } else if (fetch.getType() == SCATTERED_HASH_SHUFFLE) {
+ urlPrefix.append("s");
}
List<URI> fetchURLs = new ArrayList<URI>();
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
index ee3c40d..991dc4b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
@@ -348,6 +348,17 @@ public class Task {
Entry<Integer,String> entry = it.next();
ShuffleFileOutput.Builder part = ShuffleFileOutput.newBuilder();
part.setPartId(entry.getKey());
+
+ // Set output volume
+ if (context.getPartitionOutputVolume() != null) {
+ for (Entry<Integer, Long> e : context.getPartitionOutputVolume().entrySet()) {
+ if (entry.getKey().equals(e.getKey())) {
+ part.setVolume(e.getValue().longValue());
+ break;
+ }
+ }
+ }
+
builder.addShuffleFileOutputs(part.build());
} while (it.hasNext());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
index e073652..1f0c410 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
@@ -71,6 +71,9 @@ public class TaskAttemptContext {
private Enforcer enforcer;
private QueryContext queryContext;
+ /** a output volume for each partition */
+ private Map<Integer, Long> partitionOutputVolume;
+
public TaskAttemptContext(TajoConf conf, QueryContext queryContext, final QueryUnitAttemptId queryId,
final FragmentProto[] fragments,
final Path workDir) {
@@ -94,6 +97,8 @@ public class TaskAttemptContext {
this.shuffleFileOutputs = Maps.newHashMap();
state = TaskAttemptState.TA_PENDING;
+
+ this.partitionOutputVolume = Maps.newHashMap();
}
@VisibleForTesting
@@ -193,6 +198,19 @@ public class TaskAttemptContext {
return shuffleFileOutputs.entrySet().iterator();
}
+ public void addPartitionOutputVolume(int partId, long volume) {
+ if (partitionOutputVolume.containsKey(partId)) {
+ long sum = partitionOutputVolume.get(partId);
+ partitionOutputVolume.put(partId, sum + volume);
+ } else {
+ partitionOutputVolume.put(partId, volume);
+ }
+ }
+
+ public Map<Integer, Long> getPartitionOutputVolume() {
+ return partitionOutputVolume;
+ }
+
public void updateAssignedFragments(String tableId, Fragment[] fragments) {
fragmentMap.remove(tableId);
for(Fragment t : fragments) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/main/proto/TajoWorkerProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/proto/TajoWorkerProtocol.proto b/tajo-core/src/main/proto/TajoWorkerProtocol.proto
index 3bf6e13..ce8ce86 100644
--- a/tajo-core/src/main/proto/TajoWorkerProtocol.proto
+++ b/tajo-core/src/main/proto/TajoWorkerProtocol.proto
@@ -126,6 +126,7 @@ enum CommandType {
message ShuffleFileOutput {
required int32 partId = 1;
optional string fileName = 2;
+ optional int64 volume = 3;
}
message QueryExecutionRequestProto {
@@ -145,6 +146,7 @@ enum ShuffleType {
NONE_SHUFFLE = 0;
HASH_SHUFFLE = 1;
RANGE_SHUFFLE = 2;
+ SCATTERED_HASH_SHUFFLE = 3;
}
enum TransmitType {
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
index c34c3f4..9db8e41 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
@@ -25,21 +25,33 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.DeflateCodec;
+import org.apache.tajo.QueryId;
import org.apache.tajo.QueryTestCaseBase;
import org.apache.tajo.TajoConstants;
import org.apache.tajo.TajoTestingCluster;
import org.apache.tajo.catalog.CatalogService;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.engine.planner.global.DataChannel;
+import org.apache.tajo.engine.planner.global.ExecutionBlock;
+import org.apache.tajo.engine.planner.global.MasterPlan;
+import org.apache.tajo.engine.planner.logical.NodeType;
import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.master.querymaster.QueryMasterTask;
+import org.apache.tajo.worker.TajoWorker;
import org.junit.Test;
import java.io.IOException;
import java.sql.ResultSet;
+import java.util.List;
import java.util.Map;
+import static junit.framework.TestCase.*;
import static org.apache.tajo.TajoConstants.DEFAULT_DATABASE_NAME;
-import static org.junit.Assert.*;
+import static org.apache.tajo.ipc.TajoWorkerProtocol.ShuffleType.SCATTERED_HASH_SHUFFLE;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
public class TestTablePartitions extends QueryTestCaseBase {
@@ -62,6 +74,77 @@ public class TestTablePartitions extends QueryTestCaseBase {
res = testBase.execute(
"insert overwrite into " + tableName + " select l_orderkey, l_partkey, " +
"l_quantity from lineitem");
+
+ MasterPlan plan = getQueryPlan(res);
+ ExecutionBlock rootEB = plan.getRoot();
+
+ /*
+ -------------------------------------------------------------------------------
+ |-eb_1405354886454_0001_000003
+ |-eb_1405354886454_0001_000002
+ |-eb_1405354886454_0001_000001
+ */
+ assertEquals(1, plan.getChildCount(rootEB.getId()));
+
+ ExecutionBlock insertEB = plan.getChild(rootEB.getId(), 0);
+ assertNotNull(insertEB);
+ assertEquals(NodeType.INSERT, insertEB.getPlan().getType());
+ assertEquals(1, plan.getChildCount(insertEB.getId()));
+
+ ExecutionBlock scanEB = plan.getChild(insertEB.getId(), 0);
+
+ List<DataChannel> list = plan.getOutgoingChannels(scanEB.getId());
+ assertEquals(1, list.size());
+ DataChannel channel = list.get(0);
+ assertNotNull(channel);
+ assertEquals(SCATTERED_HASH_SHUFFLE, channel.getShuffleType());
+ assertEquals(1, channel.getShuffleKeys().length);
+
+ res.close();
+ }
+
+ @Test
+ public final void testCreateColumnPartitionedTableWithJoin() throws Exception {
+ String tableName = CatalogUtil.normalizeIdentifier("testCreateColumnPartitionedTableWithJoin");
+ ResultSet res = executeString(
+ "create table " + tableName + " (col1 int4, col2 int4) partition by column(key float8) ");
+ res.close();
+
+ assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+ assertEquals(2, catalog.getTableDesc(DEFAULT_DATABASE_NAME, tableName).getSchema().size());
+ assertEquals(3, catalog.getTableDesc(DEFAULT_DATABASE_NAME, tableName).getLogicalSchema().size());
+
+ res = testBase.execute(
+ "insert overwrite into " + tableName + " select l_orderkey, l_partkey, " +
+ "l_quantity from lineitem join orders on l_orderkey = o_orderkey");
+
+ MasterPlan plan = getQueryPlan(res);
+ ExecutionBlock rootEB = plan.getRoot();
+
+ /*
+ -------------------------------------------------------------------------------
+ |-eb_1405356074433_0001_000005
+ |-eb_1405356074433_0001_000004
+ |-eb_1405356074433_0001_000003
+ |-eb_1405356074433_0001_000002
+ |-eb_1405356074433_0001_000001
+ */
+ assertEquals(1, plan.getChildCount(rootEB.getId()));
+
+ ExecutionBlock insertEB = plan.getChild(rootEB.getId(), 0);
+ assertNotNull(insertEB);
+ assertEquals(NodeType.INSERT, insertEB.getPlan().getType());
+ assertEquals(1, plan.getChildCount(insertEB.getId()));
+
+ ExecutionBlock scanEB = plan.getChild(insertEB.getId(), 0);
+
+ List<DataChannel> list = plan.getOutgoingChannels(scanEB.getId());
+ assertEquals(1, list.size());
+ DataChannel channel = list.get(0);
+ assertNotNull(channel);
+ assertEquals(SCATTERED_HASH_SHUFFLE, channel.getShuffleType());
+ assertEquals(1, channel.getShuffleKeys().length);
+
res.close();
}
@@ -240,21 +323,13 @@ public class TestTablePartitions extends QueryTestCaseBase {
assertEquals(5, desc.getStats().getNumRows().intValue());
}
- String expected = "N\n" +
- "N\n" +
- "N\n" +
- "R\n" +
- "R\n";
-
- String tableData = getTableFileContents(desc.getPath());
- assertEquals(expected, tableData);
-
res = executeString("select * from " + tableName + " where col2 = 2");
Map<Double, int []> resultRows1 = Maps.newHashMap();
resultRows1.put(45.0d, new int[]{3, 2});
resultRows1.put(38.0d, new int[]{2, 2});
+
for (int i = 0; i < 2; i++) {
assertTrue(res.next());
assertEquals(resultRows1.get(res.getDouble(4))[0], res.getInt(2));
@@ -575,4 +650,17 @@ public class TestTablePartitions extends QueryTestCaseBase {
assertEquals(5, desc.getStats().getNumRows().intValue());
}
}
+
+ private MasterPlan getQueryPlan(ResultSet res) {
+ QueryId queryId = ((TajoResultSet)res).getQueryId();
+ for (TajoWorker eachWorker: testingCluster.getTajoWorkers()) {
+ QueryMasterTask queryMasterTask = eachWorker.getWorkerContext().getQueryMaster().getQueryMasterTask(queryId, true);
+ if (queryMasterTask != null) {
+ return queryMasterTask.getQuery().getPlan();
+ }
+ }
+
+ fail("Can't find query from workers" + queryId);
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
----------------------------------------------------------------------
diff --git a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
index b8fda29..dd3bee3 100644
--- a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
+++ b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
@@ -424,8 +424,8 @@ public class PullServerAuxService extends AuxiliaryService {
chunks.add(chunk);
}
- // if a subquery requires a hash repartition
- } else if (repartitionType.equals("h")) {
+ // if a subquery requires a hash repartition or a scattered hash repartition
+ } else if (repartitionType.equals("h") || repartitionType.equals("s")) {
for (String ta : taskIds) {
Path path = localFS.makeQualified(
lDirAlloc.getLocalPathToRead(queryBaseDir + "/" + sid + "/" +
http://git-wip-us.apache.org/repos/asf/tajo/blob/48dbfd92/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
----------------------------------------------------------------------
diff --git a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
index 373642b..5b76da5 100644
--- a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
+++ b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
@@ -430,8 +430,8 @@ public class TajoPullServerService extends AbstractService {
chunks.add(chunk);
}
- // if a subquery requires a hash shuffle
- } else if (shuffleType.equals("h")) {
+ // if a subquery requires a hash shuffle or a scattered hash shuffle
+ } else if (shuffleType.equals("h") || shuffleType.equals("s")) {
for (String ta : taskIds) {
if (!lDirAlloc.ifExists(queryBaseDir + "/" + sid + "/" + ta + "/output/" + partId, conf)) {
LOG.warn(e);
[09/19] git commit: TAJO-926: Join condition including column
references of a row-preserving table in left outer join causes incorrect
result.
Posted by ji...@apache.org.
TAJO-926: Join condition including column references of a row-preserving table in left outer join causes incorrect result.
Closes #62
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/c87fe2e8
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/c87fe2e8
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/c87fe2e8
Branch: refs/heads/index_support
Commit: c87fe2e8ca135778d3f9c91606bc41c4eb6da5a8
Parents: a17bd0f
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 18:18:54 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 18:18:54 2014 +0900
----------------------------------------------------------------------
CHANGES | 5 ++-
.../planner/physical/HashLeftOuterJoinExec.java | 41 ++++++++++++++++++--
.../planner/rewrite/ProjectionPushDownRule.java | 4 +-
.../apache/tajo/engine/query/TestJoinQuery.java | 8 ++++
.../testJoinFilterOfRowPreservedTable1.sql | 8 ++++
.../testJoinFilterOfRowPreservedTable1.result | 15 +++++++
6 files changed, 76 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/c87fe2e8/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 15bf988..2feba78 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,7 +82,10 @@ Release 0.9.0 - unreleased
BUG FIXES
- TAJO-852: Integration test using HCatalog as a catalog store is failed.
+ TAJO-926: Join condition including column references of a row-preserving
+ table in left outer join causes incorrect result. (hyunsik)
+
+ TAJO-852: Integration test using HCatalog as a catalog store is failed.
(jinho)
TAJO-934: Multiple DISTINCT returns null grouping key value.
http://git-wip-us.apache.org/repos/asf/tajo/blob/c87fe2e8/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
index 622900f..ac8b28f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
@@ -18,10 +18,13 @@
package org.apache.tajo.engine.planner.physical;
+import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tajo.catalog.Column;
+import org.apache.tajo.engine.eval.AlgebraicUtil;
import org.apache.tajo.engine.eval.EvalNode;
+import org.apache.tajo.engine.eval.EvalTreeUtil;
import org.apache.tajo.engine.planner.PlannerUtil;
import org.apache.tajo.engine.planner.Projector;
import org.apache.tajo.engine.planner.logical.JoinNode;
@@ -39,7 +42,8 @@ import java.util.*;
public class HashLeftOuterJoinExec extends BinaryPhysicalExec {
// from logical plan
protected JoinNode plan;
- protected EvalNode joinQual;
+ protected EvalNode joinQual; // ex) a.id = b.id
+ protected EvalNode joinFilter; // ex) a > 10
protected List<Column[]> joinKeyPairs;
@@ -69,7 +73,24 @@ public class HashLeftOuterJoinExec extends BinaryPhysicalExec {
super(context, SchemaUtil.merge(leftChild.getSchema(), rightChild.getSchema()),
plan.getOutSchema(), leftChild, rightChild);
this.plan = plan;
- this.joinQual = plan.getJoinQual();
+
+ List<EvalNode> joinQuals = Lists.newArrayList();
+ List<EvalNode> joinFilters = Lists.newArrayList();
+ for (EvalNode eachQual : AlgebraicUtil.toConjunctiveNormalFormArray(plan.getJoinQual())) {
+ if (EvalTreeUtil.isJoinQual(eachQual, true)) {
+ joinQuals.add(eachQual);
+ } else {
+ joinFilters.add(eachQual);
+ }
+ }
+
+ this.joinQual = AlgebraicUtil.createSingletonExprFromCNF(joinQuals.toArray(new EvalNode[joinQuals.size()]));
+ if (joinFilters.size() > 0) {
+ this.joinFilter = AlgebraicUtil.createSingletonExprFromCNF(joinFilters.toArray(new EvalNode[joinFilters.size()]));
+ } else {
+ this.joinFilter = null;
+ }
+
this.tupleSlots = new HashMap<Tuple, List<Tuple>>(10000);
// HashJoin only can manage equi join key pairs.
@@ -146,10 +167,23 @@ public class HashLeftOuterJoinExec extends BinaryPhysicalExec {
}
frameTuple.set(leftTuple, rightTuple); // evaluate a join condition on both tuples
- if (joinQual.eval(inSchema, frameTuple).isTrue()) { // if both tuples are joinable
+
+ // if there is no join filter, it is always true.
+ boolean satisfiedWithFilter = joinFilter == null ? true : joinFilter.eval(inSchema, frameTuple).isTrue();
+ boolean satisfiedWithJoinCondition = joinQual.eval(inSchema, frameTuple).isTrue();
+
+ // if a composited tuple satisfies with both join filter and join condition
+ if (satisfiedWithFilter && satisfiedWithJoinCondition) {
projector.eval(frameTuple, outTuple);
return outTuple;
} else {
+
+ // if join filter is satisfied, the left outer join (LOJ) operator should return the null padded tuple
+ // only once. Then, LOJ operator should take the next left tuple.
+ if (!satisfiedWithFilter) {
+ shouldGetLeftTuple = true;
+ }
+
// null padding
Tuple nullPaddedTuple = TupleUtil.createNullPaddedTuple(rightNumCols);
frameTuple.set(leftTuple, nullPaddedTuple);
@@ -204,6 +238,7 @@ public class HashLeftOuterJoinExec extends BinaryPhysicalExec {
iterator = null;
plan = null;
joinQual = null;
+ joinFilter = null;
projector = null;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/c87fe2e8/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
index 2bc210c..e5a329d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
@@ -834,7 +834,9 @@ public class ProjectionPushDownRule extends
// If one of both terms in a binary operator is a complex expression, the binary operator will require
// multiple phases. In this case, join cannot evaluate a binary operator.
// So, we should prevent dividing the binary operator into more subexpressions.
- if (term.getType() != EvalType.FIELD && !(term instanceof BinaryEval)) {
+ if (term.getType() != EvalType.FIELD &&
+ !(term instanceof BinaryEval) &&
+ !(term.getType() == EvalType.ROW_CONSTANT)) {
String refName = ctx.addExpr(term);
EvalTreeUtil.replace(cnf, term, new FieldEval(refName, term.getValueType()));
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/c87fe2e8/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
index ca1ece1..b59c495 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
@@ -1099,4 +1099,12 @@ public class TestJoinQuery extends QueryTestCaseBase {
executeString("DROP TABLE large_table PURGE").close();
}
}
+
+ @Test
+ public final void testJoinFilterOfRowPreservedTable1() throws Exception {
+ // this test is for join filter of a row preserved table.
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/c87fe2e8/tajo-core/src/test/resources/queries/TestJoinQuery/testJoinFilterOfRowPreservedTable1.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestJoinQuery/testJoinFilterOfRowPreservedTable1.sql b/tajo-core/src/test/resources/queries/TestJoinQuery/testJoinFilterOfRowPreservedTable1.sql
new file mode 100644
index 0000000..66274d7
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestJoinQuery/testJoinFilterOfRowPreservedTable1.sql
@@ -0,0 +1,8 @@
+select
+ r_name,
+ r_regionkey,
+ n_name,
+ n_regionkey
+from
+ region left outer join nation on n_regionkey = r_regionkey and r_name in ('AMERICA', 'ASIA')
+order by r_name;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/c87fe2e8/tajo-core/src/test/resources/results/TestJoinQuery/testJoinFilterOfRowPreservedTable1.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestJoinQuery/testJoinFilterOfRowPreservedTable1.result b/tajo-core/src/test/resources/results/TestJoinQuery/testJoinFilterOfRowPreservedTable1.result
new file mode 100644
index 0000000..82d5562
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestJoinQuery/testJoinFilterOfRowPreservedTable1.result
@@ -0,0 +1,15 @@
+r_name,r_regionkey,n_name,n_regionkey
+-------------------------------
+AFRICA,0,null,null
+AMERICA,1,ARGENTINA,1
+AMERICA,1,BRAZIL,1
+AMERICA,1,CANADA,1
+AMERICA,1,PERU,1
+AMERICA,1,UNITED STATES,1
+ASIA,2,INDIA,2
+ASIA,2,INDONESIA,2
+ASIA,2,JAPAN,2
+ASIA,2,CHINA,2
+ASIA,2,VIETNAM,2
+EUROPE,3,null,null
+MIDDLE EAST,4,null,null
\ No newline at end of file
[05/19] git commit: TAJO-927: Broadcast Join with Large, Small, Large,
Small tables makes a wrong plan. (Hyoungjun Kim via hyunsik)
Posted by ji...@apache.org.
TAJO-927: Broadcast Join with Large, Small, Large, Small tables makes a wrong plan. (Hyoungjun Kim via hyunsik)
Closes #65
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/42f3b4dd
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/42f3b4dd
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/42f3b4dd
Branch: refs/heads/index_support
Commit: 42f3b4dd8fbf1a5a46a4469c7ac919fb32089225
Parents: 438010f
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 14:48:02 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 14:48:02 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +
.../engine/planner/global/GlobalPlanner.java | 13 +--
.../planner/global/TestBroadcastJoinPlan.java | 84 ++++++++++++++++++++
3 files changed, 95 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/42f3b4dd/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index e9e512d..0694eba 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,6 +82,9 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-927: Broadcast Join with Large, Small, Large, Small tables
+ makes a wrong plan. (Hyoungjun Kim via hyunsik)
+
TAJO-913: Add some missed tests for constant value group-by keys.
(hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/42f3b4dd/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
index 2d0dd10..4e27574 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
@@ -317,14 +317,18 @@ public class GlobalPlanner {
}
JoinNode blockJoinNode = null;
- int numCandidateLargeTable = 0;
+ if (!leftBroadcast && !rightBroadcast) {
+ // In the case of large, large, small, small
+ // all small tables broadcast to right large table
+ numLargeTables = 1;
+ }
for(LogicalNode eachNode: joinNode.getBroadcastCandidateTargets()) {
- if (numCandidateLargeTable >= 2 || numLargeTables > 2) {
- break;
- }
if (eachNode.getPID() == joinNode.getPID()) {
continue;
}
+ if (numLargeTables >= 2) {
+ break;
+ }
JoinNode broadcastJoinNode = (JoinNode)eachNode;
ScanNode scanNode = broadcastJoinNode.getRightChild();
if (getTableVolume(scanNode) < broadcastThreshold) {
@@ -334,7 +338,6 @@ public class GlobalPlanner {
+ getTableVolume(scanNode) + ") is marked a broadcasted table");
} else {
numLargeTables++;
- numCandidateLargeTable++;
if (numLargeTables < 2) {
blockJoinNode = broadcastJoinNode;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/42f3b4dd/tajo-core/src/test/java/org/apache/tajo/engine/planner/global/TestBroadcastJoinPlan.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/global/TestBroadcastJoinPlan.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/global/TestBroadcastJoinPlan.java
index 7a9a1c7..fd07ae4 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/global/TestBroadcastJoinPlan.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/global/TestBroadcastJoinPlan.java
@@ -18,6 +18,7 @@
package org.apache.tajo.engine.planner.global;
+import junit.framework.TestCase;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.LocalTajoTestingUtility;
@@ -941,4 +942,87 @@ public class TestBroadcastJoinPlan {
assertEquals(3, index);
}
+
+ @Test
+ public final void testBroadcastCasebyCase1() throws IOException, PlanningException {
+ // large, small, large, small
+ String query = "select count(*) from large1 " +
+ "inner join small1 on large1_id = small1_id " +
+ "left outer join large2 on large1_id = large2_id " +
+ "left outer join small2 on large1_id = small2_id " ;
+
+ LogicalPlanner planner = new LogicalPlanner(catalog);
+ LogicalOptimizer optimizer = new LogicalOptimizer(conf);
+ Expr expr = analyzer.parse(query);
+ LogicalPlan plan = planner.createPlan(LocalTajoTestingUtility.createDummySession(), expr);
+
+ optimizer.optimize(plan);
+
+ QueryId queryId = QueryIdFactory.newQueryId(System.currentTimeMillis(), 0);
+ QueryContext queryContext = new QueryContext();
+ MasterPlan masterPlan = new MasterPlan(queryId, queryContext, plan);
+ GlobalPlanner globalPlanner = new GlobalPlanner(conf, catalog);
+ globalPlanner.build(masterPlan);
+
+ // (((default.large1 ⋈θ default.small1) ⟕ default.large2) ⟕ default.small2)
+ /*
+ |-eb_1404871198908_0000_000007
+ |-eb_1404871198908_0000_000006
+ |-eb_1404871198908_0000_000005 (join eb3, eb3, broadcast small2)
+ |-eb_1404871198908_0000_000004 (scan large2)
+ |-eb_1404871198908_0000_000003 (scan large1, broadcast small1)
+ */
+
+ ExecutionBlockCursor ebCursor = new ExecutionBlockCursor(masterPlan);
+ int index = 0;
+ while (ebCursor.hasNext()) {
+ ExecutionBlock eb = ebCursor.nextBlock();
+ if(index == 0) {
+ LogicalNode node = eb.getPlan();
+ assertEquals(NodeType.JOIN, node.getType());
+ JoinNode joinNode = (JoinNode)node;
+
+ ScanNode scanNode1 = joinNode.getLeftChild();
+ ScanNode scanNode2 = joinNode.getRightChild();
+ assertEquals("default.large1", scanNode1.getCanonicalName());
+ assertEquals("default.small1", scanNode2.getCanonicalName());
+
+ Collection<String> broadcastTables = eb.getBroadcastTables();
+
+ assertEquals(1, broadcastTables.size());
+ assertTrue(broadcastTables.contains("default.small1"));
+ } else if(index == 1) {
+ LogicalNode node = eb.getPlan();
+ assertEquals(NodeType.SCAN, node.getType());
+ ScanNode scanNode = (ScanNode)node;
+
+ assertEquals("default.large2", scanNode.getCanonicalName());
+
+ Collection<String> broadcastTables = eb.getBroadcastTables();
+ TestCase.assertEquals(0, broadcastTables.size());
+ } else if(index == 2) {
+ LogicalNode node = eb.getPlan();
+ assertEquals(NodeType.GROUP_BY, node.getType());
+ JoinNode joinNode = ((GroupbyNode)node).getChild();
+
+ JoinNode joinNode2 = joinNode.getLeftChild();
+ ScanNode scanNode = joinNode.getRightChild();
+ assertEquals("default.small2", scanNode.getCanonicalName());
+
+ ScanNode scanNode2 = joinNode2.getLeftChild();
+ ScanNode scanNode3 = joinNode2.getRightChild();
+
+ assertTrue(scanNode2.getCanonicalName().indexOf("000003") >= 0);
+ assertTrue(scanNode3.getCanonicalName().indexOf("000004") >= 0);
+
+ Collection<String> broadcastTables = eb.getBroadcastTables();
+
+ TestCase.assertEquals(1, broadcastTables.size());
+ TestCase.assertTrue(broadcastTables.contains("default.small2"));
+ }
+ index++;
+ }
+
+ TestCase.assertEquals(5, index);
+ }
}
[14/19] git commit: TAJO-932: Upgrade Parquet to 1.5.0.
Posted by ji...@apache.org.
TAJO-932: Upgrade Parquet to 1.5.0.
Closes #67
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/5e2d6c3c
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/5e2d6c3c
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/5e2d6c3c
Branch: refs/heads/index_support
Commit: 5e2d6c3c74ec107e5c62359cedb602f940061c45
Parents: fe557ad
Author: Hyunsik Choi <hy...@apache.org>
Authored: Mon Jul 14 18:36:30 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Mon Jul 14 18:36:30 2014 +0900
----------------------------------------------------------------------
CHANGES | 2 ++
tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/pom.xml | 4 ++--
tajo-storage/pom.xml | 4 ++--
3 files changed, 6 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/5e2d6c3c/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 3c326dc..c0dcbe3 100644
--- a/CHANGES
+++ b/CHANGES
@@ -27,6 +27,8 @@ Release 0.9.0 - unreleased
IMPROVEMENT
+ TAJO-932: Upgrade Parquet to 1.5.0. (hyunsik)
+
TAJO-903: Some left outer join cases are not optimized as the broadcast
join. (Hyoungjun Kim via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/5e2d6c3c/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/pom.xml b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/pom.xml
index 2c939d4..bac9d27 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/pom.xml
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/pom.xml
@@ -138,8 +138,8 @@
</activation>
<properties>
<hive.version>0.12.0</hive.version>
- <parquet.version>1.4.2</parquet.version>
- <parquet.format.version>2.0.0</parquet.format.version>
+ <parquet.version>1.5.0</parquet.version>
+ <parquet.format.version>2.1.0</parquet.format.version>
</properties>
<dependencies>
<dependency>
http://git-wip-us.apache.org/repos/asf/tajo/blob/5e2d6c3c/tajo-storage/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-storage/pom.xml b/tajo-storage/pom.xml
index 9828bce..7035b62 100644
--- a/tajo-storage/pom.xml
+++ b/tajo-storage/pom.xml
@@ -34,8 +34,8 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
- <parquet.version>1.4.2</parquet.version>
- <parquet.format.version>2.0.0</parquet.format.version>
+ <parquet.version>1.5.0</parquet.version>
+ <parquet.format.version>2.1.0</parquet.format.version>
</properties>
<repositories>
[02/19] git commit: TAJO-909: {SortBased,
Col}PartitionStoreExec should not write partition keys to files.
(Hyoungjun Kim via hyunsik)
Posted by ji...@apache.org.
TAJO-909: {SortBased, Col}PartitionStoreExec should not write partition keys to files. (Hyoungjun Kim via hyunsik)
Closes #64
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/29db645a
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/29db645a
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/29db645a
Branch: refs/heads/index_support
Commit: 29db645ad02067464ed3e9b94b3e62d44ab2adf4
Parents: 101413e
Author: Hyunsik Choi <hy...@apache.org>
Authored: Fri Jul 11 14:05:11 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Fri Jul 11 14:05:11 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 ++
.../planner/physical/ColPartitionStoreExec.java | 9 ++++-
.../java/org/apache/tajo/QueryTestCaseBase.java | 35 ++++++++++++--------
.../tajo/engine/query/TestTablePartitions.java | 10 +++++-
.../querymaster/TestQueryUnitStatusUpdate.java | 4 +--
5 files changed, 43 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/29db645a/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index c0208aa..f73bf1b 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,6 +82,9 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-909: {SortBased, Col}PartitionStoreExec should not write partition
+ keys to files. (Hyoungjun Kim via hyunsik)
+
TAJO-912: Tsql prints wrong version. (Mai Hai Thanh via hyunsik)
TAJO-902: Unicode delimiter does not work correctly. (jinho)
http://git-wip-us.apache.org/repos/asf/tajo/blob/29db645a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
index fe36905..12eb0e0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ColPartitionStoreExec.java
@@ -23,9 +23,10 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableMeta;
-import org.apache.tajo.engine.planner.logical.InsertNode;
import org.apache.tajo.engine.planner.logical.CreateTableNode;
+import org.apache.tajo.engine.planner.logical.InsertNode;
import org.apache.tajo.engine.planner.logical.NodeType;
import org.apache.tajo.engine.planner.logical.StoreTableNode;
import org.apache.tajo.storage.StorageUtil;
@@ -60,6 +61,12 @@ public abstract class ColPartitionStoreExec extends UnaryPhysicalExec {
// Find column index to name subpartition directory path
keyNum = this.plan.getPartitionMethod().getExpressionSchema().size();
+ if (plan.getType() == NodeType.INSERT && keyNum > 0) {
+ Column[] removedPartitionColumns = new Column[this.outSchema.size() - keyNum];
+ System.arraycopy(this.outSchema.toArray(), 0, removedPartitionColumns, 0, removedPartitionColumns.length);
+ this.outSchema = new Schema(removedPartitionColumns);
+ }
+
keyIds = new int[keyNum];
keyNames = new String[keyNum];
for (int i = 0; i < keyNum; i++) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/29db645a/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java b/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
index 70c73f9..29613cb 100644
--- a/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
+++ b/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
@@ -574,23 +574,14 @@ public class QueryTestCaseBase {
public String getTableFileContents(Path path) throws Exception {
FileSystem fs = path.getFileSystem(conf);
- FileStatus[] files = fs.listStatus(path);
-
- if (files == null || files.length == 0) {
- return null;
- }
-
StringBuilder sb = new StringBuilder();
- byte[] buf = new byte[1024];
-
- for (FileStatus file: files) {
- if (file.isDirectory()) {
- continue;
- }
- InputStream in = fs.open(file.getPath());
+ List<Path> paths = listFiles(fs, path);
+ for (Path eachPath: paths) {
+ InputStream in = fs.open(eachPath);
try {
while (true) {
+ byte[] buf = new byte[1024];
int readBytes = in.read(buf);
if (readBytes <= 0) {
break;
@@ -602,7 +593,6 @@ public class QueryTestCaseBase {
in.close();
}
}
-
return sb.toString();
}
@@ -621,4 +611,21 @@ public class QueryTestCaseBase {
Path path = tableDesc.getPath();
return getTableFileContents(path);
}
+
+ private List<Path> listFiles(FileSystem fs, Path path) throws Exception {
+ List<Path> result = new ArrayList<Path>();
+ FileStatus[] files = fs.listStatus(path);
+ if (files == null || files.length == 0) {
+ return result;
+ }
+
+ for (FileStatus eachFile: files) {
+ if (eachFile.isDirectory()) {
+ result.addAll(listFiles(fs, eachFile.getPath()));
+ } else {
+ result.add(eachFile.getPath());
+ }
+ }
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/29db645a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
index 8c989b5..c34c3f4 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
@@ -240,13 +240,21 @@ public class TestTablePartitions extends QueryTestCaseBase {
assertEquals(5, desc.getStats().getNumRows().intValue());
}
+ String expected = "N\n" +
+ "N\n" +
+ "N\n" +
+ "R\n" +
+ "R\n";
+
+ String tableData = getTableFileContents(desc.getPath());
+ assertEquals(expected, tableData);
+
res = executeString("select * from " + tableName + " where col2 = 2");
Map<Double, int []> resultRows1 = Maps.newHashMap();
resultRows1.put(45.0d, new int[]{3, 2});
resultRows1.put(38.0d, new int[]{2, 2});
-
for (int i = 0; i < 2; i++) {
assertTrue(res.next());
assertEquals(resultRows1.get(res.getDouble(4))[0], res.getInt(2));
http://git-wip-us.apache.org/repos/asf/tajo/blob/29db645a/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryUnitStatusUpdate.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryUnitStatusUpdate.java b/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryUnitStatusUpdate.java
index 7b9d830..e340953 100644
--- a/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryUnitStatusUpdate.java
+++ b/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryUnitStatusUpdate.java
@@ -105,8 +105,8 @@ public class TestQueryUnitStatusUpdate extends QueryTestCaseBase {
// in/out * subquery(4)
long[] expectedNumRows = new long[]{2, 2, 5, 5, 7, 2, 2, 2};
- long[] expectedNumBytes = new long[]{18, 34, 45, 75, 109, 34, 34, 18};
- long[] expectedReadBytes = new long[]{18, 0, 45, 0, 109, 0, 34, 0};
+ long[] expectedNumBytes = new long[]{8, 34, 20, 75, 109, 34, 34, 18};
+ long[] expectedReadBytes = new long[]{8, 0, 20, 0, 109, 0, 34, 0};
assertStatus(4, expectedNumRows, expectedNumBytes, expectedReadBytes);
} finally {
[12/19] git commit: TAJO-869: Sometimes,
the unit test of testTaskRunnerHistory is failed. (jinho)
Posted by ji...@apache.org.
TAJO-869: Sometimes, the unit test of testTaskRunnerHistory is failed. (jinho)
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/0a39818f
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/0a39818f
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/0a39818f
Branch: refs/heads/index_support
Commit: 0a39818f5ee29be56b23060c978915fd7287fd0a
Parents: 01857da
Author: jinossy <ji...@gmail.com>
Authored: Sat Jul 12 00:28:31 2014 +0900
Committer: jinossy <ji...@gmail.com>
Committed: Sat Jul 12 00:28:31 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +++
tajo-core/src/main/java/org/apache/tajo/worker/Task.java | 8 +++-----
.../main/java/org/apache/tajo/worker/TaskRunnerHistory.java | 5 +++--
3 files changed, 9 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/0a39818f/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 87cf7b0..26d3341 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,6 +82,9 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-869: Sometimes, the unit test of testTaskRunnerHistory is failed.
+ (jinho)
+
TAJO-908: Fetcher does not retry, when pull server connection was closed.
(jinho)
http://git-wip-us.apache.org/repos/asf/tajo/blob/0a39818f/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
index 9350838..ee3c40d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
@@ -85,7 +85,6 @@ public class Task {
private boolean interQuery;
private boolean killed = false;
private boolean aborted = false;
- private boolean stopped = false;
private final Reporter reporter;
private Path inputTableBaseDir;
@@ -411,7 +410,6 @@ public class Task {
aborted = true;
} finally {
context.setProgress(1.0f);
- stopped = true;
taskRunnerContext.completedTasksNum.incrementAndGet();
if (killed || aborted) {
@@ -466,7 +464,7 @@ public class Task {
finishTime = System.currentTimeMillis();
LOG.info("Worker's task counter - total:" + taskRunnerContext.completedTasksNum.intValue() +
", succeeded: " + taskRunnerContext.succeededTasksNum.intValue()
- + ", killed: " + taskRunnerContext.killedTasksNum.incrementAndGet()
+ + ", killed: " + taskRunnerContext.killedTasksNum.intValue()
+ ", failed: " + taskRunnerContext.failedTasksNum.intValue());
cleanupTask();
}
@@ -711,7 +709,7 @@ public class Task {
int remainingRetries = MAX_RETRIES;
@Override
public void run() {
- while (!stop.get() && !stopped) {
+ while (!stop.get() && !context.isStopped()) {
try {
if(executor != null && context.getProgress() < 1.0f) {
float progress = executor.getProgress();
@@ -736,7 +734,7 @@ public class Task {
throw new RuntimeException(t);
}
} finally {
- if (remainingRetries > 0) {
+ if (!context.isStopped() && remainingRetries > 0) {
synchronized (pingThread) {
try {
pingThread.wait(PROGRESS_INTERVAL);
http://git-wip-us.apache.org/repos/asf/tajo/blob/0a39818f/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerHistory.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerHistory.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerHistory.java
index df60855..a8a11c1 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerHistory.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerHistory.java
@@ -57,7 +57,7 @@ public class TaskRunnerHistory implements ProtoObject<TaskRunnerHistoryProto> {
this.startTime = proto.getStartTime();
this.finishTime = proto.getFinishTime();
this.executionBlockId = new ExecutionBlockId(proto.getExecutionBlockId());
- this.taskHistoryMap = Maps.newHashMap();
+ this.taskHistoryMap = Maps.newTreeMap();
for (TaskHistoryProto taskHistoryProto : proto.getTaskHistoriesList()) {
TaskHistory taskHistory = new TaskHistory(taskHistoryProto);
taskHistoryMap.put(taskHistory.getQueryUnitAttemptId(), taskHistory);
@@ -74,7 +74,8 @@ public class TaskRunnerHistory implements ProtoObject<TaskRunnerHistoryProto> {
@Override
public int hashCode() {
- return Objects.hashCode(containerId, executionBlockId, taskHistoryMap);
+ return Objects.hashCode(containerId, executionBlockId, state, startTime,
+ finishTime, taskHistoryMap.size());
}
@Override
[17/19] git commit: TAJO-904: ORDER BY Null first support. (Hyoungjun
Kim via hyunsik)
Posted by ji...@apache.org.
TAJO-904: ORDER BY Null first support. (Hyoungjun Kim via hyunsik)
Closes #70
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/47d4fe22
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/47d4fe22
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/47d4fe22
Branch: refs/heads/index_support
Commit: 47d4fe22b8e5594d82054fa5a7b5cdc23f578be6
Parents: aee7874
Author: Hyunsik Choi <hy...@apache.org>
Authored: Tue Jul 15 14:55:47 2014 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Tue Jul 15 14:55:47 2014 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../tajo/catalog/statistics/ColumnStats.java | 10 ++-
.../engine/planner/RangePartitionAlgorithm.java | 2 +-
.../org/apache/tajo/engine/utils/TupleUtil.java | 70 +++++++++++++++++---
.../tajo/master/querymaster/Repartitioner.java | 19 +++++-
.../main/java/org/apache/tajo/worker/Task.java | 4 +-
.../apache/tajo/engine/query/TestSortQuery.java | 49 ++++++++++++++
.../org/apache/tajo/storage/RowStoreUtil.java | 1 +
.../apache/tajo/storage/TableStatistics.java | 5 +-
.../tajo/pullserver/TajoPullServerService.java | 2 +-
10 files changed, 141 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 473db19..d819d20 100644
--- a/CHANGES
+++ b/CHANGES
@@ -84,6 +84,8 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-904: ORDER BY Null first support. (Hyoungjun Kim via hyunsik)
+
TAJO-936: TestStorages::testSplitable is failed occasionally. (jinho)
TAJO-673: Assign proper number of tasks when inserting into partitioned table. (jaehwa)
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/ColumnStats.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/ColumnStats.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/ColumnStats.java
index 4d65d9a..1f3bca3 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/ColumnStats.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/statistics/ColumnStats.java
@@ -24,13 +24,13 @@ package org.apache.tajo.catalog.statistics;
import com.google.common.base.Objects;
import com.google.gson.annotations.Expose;
import com.google.protobuf.ByteString;
-import org.apache.tajo.catalog.proto.CatalogProtos;
-import org.apache.tajo.datum.DatumFactory;
-import org.apache.tajo.json.GsonObject;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.json.CatalogGsonHelper;
+import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.common.ProtoObject;
import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.json.GsonObject;
import org.apache.tajo.util.TUtil;
public class ColumnStats implements ProtoObject<CatalogProtos.ColumnStatsProto>, Cloneable, GsonObject {
@@ -109,6 +109,10 @@ public class ColumnStats implements ProtoObject<CatalogProtos.ColumnStatsProto>,
this.numNulls = numNulls;
}
+ public boolean hasNullValue() {
+ return numNulls > 0;
+ }
+
public boolean equals(Object obj) {
if (obj instanceof ColumnStats) {
ColumnStats other = (ColumnStats) obj;
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java
index a3522c7..0aa6f97 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/RangePartitionAlgorithm.java
@@ -115,7 +115,7 @@ public abstract class RangePartitionAlgorithm {
break;
case TEXT:
final char textStart = (start instanceof NullDatum || start.size() == 0) ? '0' : start.asChars().charAt(0);
- final char textEnd = (end instanceof NullDatum || start.size() == 0) ? '0' : end.asChars().charAt(0);
+ final char textEnd = (end instanceof NullDatum || end.size() == 0) ? '0' : end.asChars().charAt(0);
if (isAscending) {
columnCard = new BigDecimal(textEnd - textStart);
} else {
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleUtil.java b/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleUtil.java
index 86f4935..f2e47bc 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleUtil.java
@@ -22,6 +22,8 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.Schema;
@@ -44,6 +46,7 @@ import java.util.List;
import java.util.Map;
public class TupleUtil {
+ private static final Log LOG = LogFactory.getLog(TupleUtil.class);
public static String rangeToQuery(Schema schema, TupleRange range, boolean last)
throws UnsupportedEncodingException {
@@ -72,7 +75,41 @@ public class TupleUtil {
return sb.toString();
}
- public static TupleRange columnStatToRange(SortSpec [] sortSpecs, Schema target, List<ColumnStats> colStats) {
+ /**
+ * if max value is null, set ranges[last]
+ * @param sortSpecs
+ * @param sortSchema
+ * @param colStats
+ * @param ranges
+ */
+ public static void setMaxRangeIfNull(SortSpec[] sortSpecs, Schema sortSchema,
+ List<ColumnStats> colStats, TupleRange[] ranges) {
+ Map<Column, ColumnStats> statMap = Maps.newHashMap();
+ for (ColumnStats stat : colStats) {
+ statMap.put(stat.getColumn(), stat);
+ }
+
+ int i = 0;
+ for (Column col : sortSchema.getColumns()) {
+ ColumnStats columnStat = statMap.get(col);
+ if (columnStat == null) {
+ continue;
+ }
+ if (columnStat.hasNullValue()) {
+ int rangeIndex = sortSpecs[i].isAscending() ? ranges.length - 1 : 0;
+ VTuple rangeTuple = sortSpecs[i].isAscending() ? (VTuple) ranges[rangeIndex].getEnd() : (VTuple) ranges[rangeIndex].getStart();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Set null into range: " + col.getQualifiedName() + ", previous tuple is " + rangeTuple);
+ }
+ rangeTuple.put(i, NullDatum.get());
+ LOG.info("Set null into range: " + col.getQualifiedName() + ", current tuple is " + rangeTuple);
+ }
+ i++;
+ }
+ }
+
+ public static TupleRange columnStatToRange(SortSpec [] sortSpecs, Schema target, List<ColumnStats> colStats,
+ boolean checkNull) {
Map<Column, ColumnStats> statSet = Maps.newHashMap();
for (ColumnStats stat : colStats) {
@@ -98,16 +135,29 @@ public class TupleUtil {
else
startTuple.put(i, DatumFactory.createNullDatum());
- if (statSet.get(col).getMaxValue() != null)
- endTuple.put(i, statSet.get(col).getMaxValue());
- else
- endTuple.put(i, DatumFactory.createNullDatum());
+ if (checkNull) {
+ if (statSet.get(col).hasNullValue() || statSet.get(col).getMaxValue() == null)
+ endTuple.put(i, DatumFactory.createNullDatum());
+ else
+ endTuple.put(i, statSet.get(col).getMaxValue());
+ } else {
+ if (statSet.get(col).getMaxValue() != null)
+ endTuple.put(i, statSet.get(col).getMaxValue());
+ else
+ endTuple.put(i, DatumFactory.createNullDatum());
+ }
} else {
- if (statSet.get(col).getMaxValue() != null)
- startTuple.put(i, statSet.get(col).getMaxValue());
- else
- startTuple.put(i, DatumFactory.createNullDatum());
-
+ if (checkNull) {
+ if (statSet.get(col).hasNullValue() || statSet.get(col).getMaxValue() == null)
+ startTuple.put(i, DatumFactory.createNullDatum());
+ else
+ startTuple.put(i, statSet.get(col).getMaxValue());
+ } else {
+ if (statSet.get(col).getMaxValue() != null)
+ startTuple.put(i, statSet.get(col).getMaxValue());
+ else
+ startTuple.put(i, DatumFactory.createNullDatum());
+ }
if (statSet.get(col).getMinValue() != null)
endTuple.put(i, statSet.get(col).getMinValue());
else
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
index 1cc5b78..055e9a2 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java
@@ -572,7 +572,7 @@ public class Repartitioner {
if (totalStat.getNumBytes() == 0 && totalStat.getColumnStats().size() == 0 ) {
return;
}
- TupleRange mergedRange = TupleUtil.columnStatToRange(sortSpecs, sortSchema, totalStat.getColumnStats());
+ TupleRange mergedRange = TupleUtil.columnStatToRange(sortSpecs, sortSchema, totalStat.getColumnStats(), false);
RangePartitionAlgorithm partitioner = new UniformRangePartition(mergedRange, sortSpecs);
BigDecimal card = partitioner.getTotalCardinality();
@@ -580,16 +580,29 @@ public class Repartitioner {
// we set the the number of tasks to the number of range cardinality.
int determinedTaskNum;
if (card.compareTo(new BigDecimal(maxNum)) < 0) {
- LOG.info("The range cardinality (" + card
+ LOG.info(subQuery.getId() + ", The range cardinality (" + card
+ ") is less then the desired number of tasks (" + maxNum + ")");
determinedTaskNum = card.intValue();
} else {
determinedTaskNum = maxNum;
}
- LOG.info("Try to divide " + mergedRange + " into " + determinedTaskNum +
+ // for LOG
+ TupleRange mergedRangeForPrint = TupleUtil.columnStatToRange(sortSpecs, sortSchema, totalStat.getColumnStats(), true);
+ LOG.info(subQuery.getId() + ", Try to divide " + mergedRangeForPrint + " into " + determinedTaskNum +
" sub ranges (total units: " + determinedTaskNum + ")");
TupleRange [] ranges = partitioner.partition(determinedTaskNum);
+ if (ranges == null || ranges.length == 0) {
+ LOG.warn(subQuery.getId() + " no range infos.");
+ }
+ TupleUtil.setMaxRangeIfNull(sortSpecs, sortSchema, totalStat.getColumnStats(), ranges);
+ if (LOG.isDebugEnabled()) {
+ if (ranges != null) {
+ for (TupleRange eachRange : ranges) {
+ LOG.debug(subQuery.getId() + " range: " + eachRange.getStart() + " ~ " + eachRange.getEnd());
+ }
+ }
+ }
FileFragment dummyFragment = new FileFragment(scan.getTableName(), tablePath, 0, 0, new String[]{UNKNOWN_HOST});
SubQuery.scheduleFragment(subQuery, dummyFragment);
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
index 991dc4b..c3f3827 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
@@ -342,10 +342,10 @@ public class Task {
builder.setResultStats(new TableStats().getProto());
}
- Iterator<Entry<Integer,String>> it = context.getShuffleFileOutputs();
+ Iterator<Entry<Integer, String>> it = context.getShuffleFileOutputs();
if (it.hasNext()) {
do {
- Entry<Integer,String> entry = it.next();
+ Entry<Integer, String> entry = it.next();
ShuffleFileOutput.Builder part = ShuffleFileOutput.newBuilder();
part.setPartId(entry.getKey());
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
index 0b1831c..a520e56 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
@@ -21,13 +21,21 @@ package org.apache.tajo.engine.query;
import org.apache.tajo.IntegrationTest;
import org.apache.tajo.QueryTestCaseBase;
import org.apache.tajo.TajoConstants;
+import org.apache.tajo.TajoTestingCluster;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.storage.StorageConstants;
+import org.apache.tajo.util.KeyValueSet;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import java.sql.ResultSet;
import java.util.TimeZone;
+import static org.junit.Assert.assertEquals;
+
@Category(IntegrationTest.class)
public class TestSortQuery extends QueryTestCaseBase {
@@ -169,4 +177,45 @@ public class TestSortQuery extends QueryTestCaseBase {
assertResultSet(res);
cleanupQuery(res);
}
+
+ @Test
+ public final void testSortNullColumn() throws Exception {
+ try {
+ testingCluster.setAllTajoDaemonConfValue(ConfVars.TESTCASE_MIN_TASK_NUM.varname, "2");
+ KeyValueSet tableOptions = new KeyValueSet();
+ tableOptions.put(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.put(StorageConstants.CSVFILE_NULL, "\\\\N");
+
+ Schema schema = new Schema();
+ schema.addColumn("id", Type.INT4);
+ schema.addColumn("name", Type.TEXT);
+ String[] data = new String[]{
+ "1|BRAZIL",
+ "2|ALGERIA",
+ "3|ARGENTINA",
+ "4|CANADA"
+ };
+ TajoTestingCluster.createTable("nullsort", schema, tableOptions, data, 2);
+
+ ResultSet res = executeString(
+ "select * from (" +
+ "select case when id > 2 then null else id end as col1, name as col2 from nullsort) a " +
+ "order by col1, col2"
+ );
+
+ String expected = "col1,col2\n" +
+ "-------------------------------\n" +
+ "1,BRAZIL\n" +
+ "2,ALGERIA\n" +
+ "null,ARGENTINA\n" +
+ "null,CANADA\n";
+
+ assertEquals(expected, resultSetToString(res));
+
+ cleanupQuery(res);
+ } finally {
+ testingCluster.setAllTajoDaemonConfValue(ConfVars.TESTCASE_MIN_TASK_NUM.varname, "0");
+ executeString("DROP TABLE nullsort PURGE;").close();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java b/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
index 33b2ff3..5140a63 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
@@ -181,6 +181,7 @@ public class RowStoreUtil {
for (int i = 0; i < schema.size(); i++) {
if (tuple.isNull(i)) {
nullFlags.set(i);
+ continue;
}
col = schema.getColumn(i);
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-storage/src/main/java/org/apache/tajo/storage/TableStatistics.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/TableStatistics.java b/tajo-storage/src/main/java/org/apache/tajo/storage/TableStatistics.java
index ac9bd8a..a2c08de 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/TableStatistics.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/TableStatistics.java
@@ -40,7 +40,6 @@ public class TableStatistics {
private long numRows = 0;
private long numBytes = 0;
-
private boolean [] comparable;
public TableStatistics(Schema schema) {
@@ -113,10 +112,10 @@ public class TableStatistics {
LOG.warn("Wrong statistics column type (" + minValues.get(i).type() +
", expected=" + schema.getColumn(i).getDataType().getType() + ")");
}
- if (minValues.get(i) == null || schema.getColumn(i).getDataType().getType() == minValues.get(i).type()) {
+ if (maxValues.get(i) == null || schema.getColumn(i).getDataType().getType() == maxValues.get(i).type()) {
columnStats.setMaxValue(maxValues.get(i));
} else {
- LOG.warn("Wrong statistics column type (" + minValues.get(i).type() +
+ LOG.warn("Wrong statistics column type (" + maxValues.get(i).type() +
", expected=" + schema.getColumn(i).getDataType().getType() + ")");
}
stat.addColumnStat(columnStats);
http://git-wip-us.apache.org/repos/asf/tajo/blob/47d4fe22/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
----------------------------------------------------------------------
diff --git a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
index 5b76da5..12cd1a3 100644
--- a/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
+++ b/tajo-yarn-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
@@ -597,7 +597,7 @@ public class TajoPullServerService extends AbstractService {
if (comparator.compare(end, idxReader.getFirstKey()) < 0 ||
comparator.compare(idxReader.getLastKey(), start) < 0) {
- LOG.info("Out of Scope (indexed data [" + idxReader.getFirstKey() + ", " + idxReader.getLastKey() +
+ LOG.warn("Out of Scope (indexed data [" + idxReader.getFirstKey() + ", " + idxReader.getLastKey() +
"], but request start:" + start + ", end: " + end);
return null;
}
[08/19] git commit: TAJO-852: Integration test using HCatalog as a
catalog store is failed. (jinho)
Posted by ji...@apache.org.
TAJO-852: Integration test using HCatalog as a catalog store is failed. (jinho)
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/a17bd0f1
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/a17bd0f1
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/a17bd0f1
Branch: refs/heads/index_support
Commit: a17bd0f182d5405450b3e47be81f45dd6cdc362e
Parents: 95292d2
Author: jinossy <ji...@gmail.com>
Authored: Fri Jul 11 17:05:38 2014 +0900
Committer: jinossy <ji...@gmail.com>
Committed: Fri Jul 11 17:05:38 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +
.../tajo/catalog/store/HCatalogStore.java | 25 ++-
.../catalog/store/HCatalogStoreClientPool.java | 32 ++-
.../java/org/apache/tajo/util/KeyValueSet.java | 2 +-
.../tajo/engine/query/TestCreateTable.java | 199 +++++++++++++------
.../org/apache/tajo/jdbc/TestResultSet.java | 3 +
.../java/org/apache/tajo/rpc/TestAsyncRpc.java | 22 +-
7 files changed, 182 insertions(+), 104 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/a17bd0f1/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 43075f0..15bf988 100644
--- a/CHANGES
+++ b/CHANGES
@@ -82,6 +82,9 @@ Release 0.9.0 - unreleased
BUG FIXES
+ TAJO-852: Integration test using HCatalog as a catalog store is failed.
+ (jinho)
+
TAJO-934: Multiple DISTINCT returns null grouping key value.
(Hyoungjun Kim via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/a17bd0f1/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
index 3008ed9..6f48348 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
@@ -82,17 +82,15 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
// get table
try {
- try {
- client = clientPool.getClient();
- table = HCatUtil.getTable(client.getHiveClient(), databaseName, tableName);
- if (table != null) {
- exist = true;
- }
- } catch (NoSuchObjectException nsoe) {
- exist = false;
- } catch (Exception e) {
- throw new CatalogException(e);
+ client = clientPool.getClient();
+ table = HCatUtil.getTable(client.getHiveClient(), databaseName, tableName);
+ if (table != null) {
+ exist = true;
}
+ } catch (NoSuchObjectException nsoe) {
+ exist = false;
+ } catch (Exception e) {
+ throw new CatalogException(e);
} finally {
if (client != null) {
client.release();
@@ -168,6 +166,8 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
stats = new TableStats();
options = new KeyValueSet();
options.putAll(table.getParameters());
+ options.delete("EXTERNAL");
+
Properties properties = table.getMetadata();
if (properties != null) {
// set field delimiter
@@ -186,6 +186,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
} else {
nullFormat = "\\N";
}
+ options.delete(serdeConstants.SERIALIZATION_NULL_FORMAT);
// set file output format
String fileOutputformat = properties.getProperty(hive_metastoreConstants.FILE_OUTPUT_FORMAT);
@@ -260,8 +261,10 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
if(client != null) client.release();
}
TableMeta meta = new TableMeta(storeType, options);
-
TableDesc tableDesc = new TableDesc(databaseName + "." + tableName, schema, meta, path);
+ if (table.getTableType().equals(TableType.EXTERNAL_TABLE)) {
+ tableDesc.setExternal(true);
+ }
if (stats != null) {
tableDesc.setStats(stats);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a17bd0f1/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
index 5c416c6..2fff67c 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStoreClientPool.java
@@ -14,7 +14,6 @@
package org.apache.tajo.catalog.store;
-import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
@@ -43,7 +42,7 @@ public class HCatalogStoreClientPool {
*/
public class HCatalogStoreClient {
private final HiveMetaStoreClient hiveClient;
- private boolean isInUse;
+ public AtomicBoolean isInUse = new AtomicBoolean(false);
private HCatalogStoreClient(HiveConf hiveConf) {
try {
@@ -54,7 +53,6 @@ public class HCatalogStoreClientPool {
// Turn in to an unchecked exception
throw new IllegalStateException(e);
}
- this.isInUse = false;
}
/**
@@ -68,24 +66,23 @@ public class HCatalogStoreClientPool {
* Returns this client back to the connection pool. If the connection pool has been
* closed, just close the Hive client connection.
*/
- public void release() {
- Preconditions.checkState(isInUse);
- isInUse = false;
+ public synchronized void release() {
+ if(!this.isInUse.getAndSet(false)){
+ return;
+ }
// Ensure the connection isn't returned to the pool if the pool has been closed.
// This lock is needed to ensure proper behavior when a thread reads poolClosed
// is false, but a call to pool.close() comes in immediately afterward.
- synchronized (poolClosed) {
- if (poolClosed.get()) {
- this.getHiveClient().close();
- } else {
- clientPool.add(this);
- }
+ if (poolClosed.get()) {
+ this.getHiveClient().close();
+ } else {
+ clientPool.add(this);
}
}
// Marks this client as in use
private void markInUse() {
- isInUse = true;
+ isInUse.set(true);
}
}
@@ -123,7 +120,7 @@ public class HCatalogStoreClientPool {
/**
* Gets a client from the pool. If the pool is empty a new client is created.
*/
- public HCatalogStoreClient getClient() {
+ public synchronized HCatalogStoreClient getClient() {
// The MetaStoreClient c'tor relies on knowing the Hadoop version by asking
// org.apache.hadoop.util.VersionInfo. The VersionInfo class relies on opening
// the 'common-version-info.properties' file as a resource from hadoop-common*.jar
@@ -149,11 +146,8 @@ public class HCatalogStoreClientPool {
*/
public void close() {
// Ensure no more items get added to the pool once close is called.
- synchronized (poolClosed) {
- if (poolClosed.get()) {
- return;
- }
- poolClosed.set(true);
+ if (poolClosed.getAndSet(true)) {
+ return;
}
HCatalogStoreClient client = null;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a17bd0f1/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java b/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
index 396c41e..4d1cee1 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
@@ -111,7 +111,7 @@ public class KeyValueSet implements ProtoObject<KeyValueSetProto>, Cloneable, Gs
if(object instanceof KeyValueSet) {
KeyValueSet other = (KeyValueSet)object;
for(Entry<String, String> entry : other.keyVals.entrySet()) {
- if(!keyVals.get(entry.getKey()).equals(entry.getValue()))
+ if(!keyVals.containsKey(entry.getKey()) || !keyVals.get(entry.getKey()).equals(entry.getValue()))
return false;
}
return true;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a17bd0f1/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCreateTable.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCreateTable.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCreateTable.java
index 3d90a79..cb77936 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCreateTable.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCreateTable.java
@@ -441,71 +441,139 @@ public class TestCreateTable extends QueryTestCaseBase {
@Test
public final void testCreateTableLike1() throws Exception {
- // Basic create table with default database
- executeString("CREATE TABLE table1 (c1 int, c2 varchar);").close();
- executeString("CREATE TABLE table2 LIKE table1");
- String testMsg = "testCreateTableLike1: Basic create table with default db";
- assertTrue(testMsg,isClonedTable("table1","table2"));
- executeString("DROP TABLE table1");
- executeString("DROP TABLE table2");
-
- // Basic create table with database
- executeString("CREATE DATABASE d1").close();
- executeString("CREATE TABLE d1.table1 (c1 int, c2 varchar);").close();
- executeString("CREATE TABLE d1.table2 LIKE d1.table1");
- testMsg = "testCreateTableLike1: Basic create table with db test failed";
- assertTrue(testMsg, isClonedTable("d1.table1","d1.table2"));
- executeString("DROP TABLE d1.table1");
- executeString("DROP TABLE d1.table2");
-
- // Table with non-default store type
- executeString("CREATE TABLE table1 (c1 int, c2 varchar) USING rcfile;").close();
- executeString("CREATE TABLE table2 LIKE table1");
- testMsg = "testCreateTableLike1: Table with non-default store type test failed";
- assertTrue(testMsg, isClonedTable("table1","table2"));
- executeString("DROP TABLE table1");
- executeString("DROP TABLE table2");
-
- // Table with non-default meta options
- executeString("CREATE TABLE table1 (c1 int, c2 varchar) USING csv WITH ('csvfile.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec');").close();
- executeString("CREATE TABLE table2 LIKE table1");
- testMsg = "testCreateTableLike1: Table with non-default meta options test failed";
- assertTrue(testMsg, isClonedTable("table1","table2"));
- executeString("DROP TABLE table1");
- executeString("DROP TABLE table2");
-
-
- // Table with partitions (default partition type)
- executeString("CREATE TABLE table1 (c1 int, c2 varchar) PARTITION BY COLUMN (c3 int, c4 float, c5 text);").close();
- executeString("CREATE TABLE table2 LIKE table1");
- testMsg = "testCreateTableLike1: Table with partitions test failed";
- assertTrue(testMsg, isClonedTable("table1","table2"));
- executeString("DROP TABLE table1");
- executeString("DROP TABLE table2");
-
-
- // Table with external flag
- // Use existing file as input for creating external table
- String className = getClass().getSimpleName();
- Path currentDatasetPath = new Path(datasetBasePath, className);
- Path filePath = StorageUtil.concatPath(currentDatasetPath, "table1");
- executeString("CREATE EXTERNAL TABLE table3 (c1 int, c2 varchar) USING rcfile LOCATION '" + filePath.toUri() + "'").close();
- executeString("CREATE TABLE table2 LIKE table3");
- testMsg = "testCreateTableLike1: Table with external table flag test failed";
- assertTrue(testMsg, isClonedTable("table3","table2"));
- executeString("DROP TABLE table3");
- executeString("DROP TABLE table2");
-
-
- // Table created using CTAS
- executeString("CREATE TABLE table3 (c1 int, c2 varchar) PARTITION BY COLUMN (c3 int);").close();
- executeString("CREATE TABLE table4 AS SELECT c1*c1, c2, c2 as c2_a,c3 from table3;").close();
- executeString("CREATE TABLE table2 LIKE table4");
- testMsg = "testCreateTableLike1: Table using CTAS test failed";
- assertTrue(testMsg, isClonedTable("table4","table2"));
- executeString("DROP TABLE table3");
- executeString("DROP TABLE table4");
- executeString("DROP TABLE table2");
+ // Hcatalog does not support varchar type in hive-0.12.0
+ if (testingCluster.isHCatalogStoreRunning()) {
+ // Basic create table with default database
+ executeString("CREATE TABLE table1 (c1 int, c2 text);").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ String testMsg = "testCreateTableLike1: Basic create table with default db";
+ assertTrue(testMsg,isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+ // Basic create table with database
+ executeString("CREATE DATABASE d1").close();
+ executeString("CREATE TABLE d1.table1 (c1 int, c2 text);").close();
+ executeString("CREATE TABLE d1.table2 LIKE d1.table1");
+ testMsg = "testCreateTableLike1: Basic create table with db test failed";
+ assertTrue(testMsg, isClonedTable("d1.table1","d1.table2"));
+ executeString("DROP TABLE d1.table1");
+ executeString("DROP TABLE d1.table2");
+
+ // Table with non-default store type
+ executeString("CREATE TABLE table1 (c1 int, c2 text) USING rcfile;").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ testMsg = "testCreateTableLike1: Table with non-default store type test failed";
+ assertTrue(testMsg, isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+ // Table with non-default meta options
+ executeString("CREATE TABLE table1 (c1 int, c2 text) USING csv WITH ('csvfile.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec');").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ testMsg = "testCreateTableLike1: Table with non-default meta options test failed";
+ assertTrue(testMsg, isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+
+ // Table with partitions (default partition type)
+ executeString("CREATE TABLE table1 (c1 int, c2 text) PARTITION BY COLUMN (c3 int, c4 float, c5 text);").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ testMsg = "testCreateTableLike1: Table with partitions test failed";
+ assertTrue(testMsg, isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+
+ // Table with external flag
+ // Use existing file as input for creating external table
+ String className = getClass().getSimpleName();
+ Path currentDatasetPath = new Path(datasetBasePath, className);
+ Path filePath = StorageUtil.concatPath(currentDatasetPath, "table1");
+ executeString("CREATE EXTERNAL TABLE table3 (c1 int, c2 text) USING rcfile LOCATION '" + filePath.toUri() + "'").close();
+ executeString("CREATE TABLE table2 LIKE table3");
+ testMsg = "testCreateTableLike1: Table with external table flag test failed";
+ assertTrue(testMsg, isClonedTable("table3","table2"));
+ executeString("DROP TABLE table3");
+ executeString("DROP TABLE table2");
+
+
+ // Table created using CTAS
+ executeString("CREATE TABLE table3 (c1 int, c2 text) PARTITION BY COLUMN (c3 int);").close();
+ executeString("CREATE TABLE table4 AS SELECT c1 * c1 as m_c1, c2, c2 as c2_a,c3 from table3;").close();
+ executeString("CREATE TABLE table2 LIKE table4");
+ testMsg = "testCreateTableLike1: Table using CTAS test failed";
+ assertTrue(testMsg, isClonedTable("table4","table2"));
+ executeString("DROP TABLE table3");
+ executeString("DROP TABLE table4");
+ executeString("DROP TABLE table2");
+ } else {
+ // Basic create table with default database
+ executeString("CREATE TABLE table1 (c1 int, c2 varchar);").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ String testMsg = "testCreateTableLike1: Basic create table with default db";
+ assertTrue(testMsg,isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+ // Basic create table with database
+ executeString("CREATE DATABASE d1").close();
+ executeString("CREATE TABLE d1.table1 (c1 int, c2 varchar);").close();
+ executeString("CREATE TABLE d1.table2 LIKE d1.table1");
+ testMsg = "testCreateTableLike1: Basic create table with db test failed";
+ assertTrue(testMsg, isClonedTable("d1.table1","d1.table2"));
+ executeString("DROP TABLE d1.table1");
+ executeString("DROP TABLE d1.table2");
+
+ // Table with non-default store type
+ executeString("CREATE TABLE table1 (c1 int, c2 varchar) USING rcfile;").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ testMsg = "testCreateTableLike1: Table with non-default store type test failed";
+ assertTrue(testMsg, isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+ // Table with non-default meta options
+ executeString("CREATE TABLE table1 (c1 int, c2 varchar) USING csv WITH ('csvfile.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec');").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ testMsg = "testCreateTableLike1: Table with non-default meta options test failed";
+ assertTrue(testMsg, isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+
+ // Table with partitions (default partition type)
+ executeString("CREATE TABLE table1 (c1 int, c2 varchar) PARTITION BY COLUMN (c3 int, c4 float, c5 text);").close();
+ executeString("CREATE TABLE table2 LIKE table1");
+ testMsg = "testCreateTableLike1: Table with partitions test failed";
+ assertTrue(testMsg, isClonedTable("table1","table2"));
+ executeString("DROP TABLE table1");
+ executeString("DROP TABLE table2");
+
+
+ // Table with external flag
+ // Use existing file as input for creating external table
+ String className = getClass().getSimpleName();
+ Path currentDatasetPath = new Path(datasetBasePath, className);
+ Path filePath = StorageUtil.concatPath(currentDatasetPath, "table1");
+ executeString("CREATE EXTERNAL TABLE table3 (c1 int, c2 varchar) USING rcfile LOCATION '" + filePath.toUri() + "'").close();
+ executeString("CREATE TABLE table2 LIKE table3");
+ testMsg = "testCreateTableLike1: Table with external table flag test failed";
+ assertTrue(testMsg, isClonedTable("table3","table2"));
+ executeString("DROP TABLE table3");
+ executeString("DROP TABLE table2");
+
+
+ // Table created using CTAS
+ executeString("CREATE TABLE table3 (c1 int, c2 varchar) PARTITION BY COLUMN (c3 int);").close();
+ executeString("CREATE TABLE table4 AS SELECT c1*c1, c2, c2 as c2_a,c3 from table3;").close();
+ executeString("CREATE TABLE table2 LIKE table4");
+ testMsg = "testCreateTableLike1: Table using CTAS test failed";
+ assertTrue(testMsg, isClonedTable("table4","table2"));
+ executeString("DROP TABLE table3");
+ executeString("DROP TABLE table4");
+ executeString("DROP TABLE table2");
/* Enable when view is supported
@@ -529,5 +597,6 @@ public class TestCreateTable extends QueryTestCaseBase {
executeString("DROP TABLE table1");
executeString("DROP TABLE table2");
*/
+ }
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a17bd0f1/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java b/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
index fdbca61..7ca94c9 100644
--- a/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
+++ b/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
@@ -129,6 +129,9 @@ public class TestResultSet {
@Test
public void testDateTimeType() throws Exception {
+ // Hcatalog does not support date type, time type in hive-0.12.0
+ if(util.isHCatalogStoreRunning()) return;
+
TimeZone tajoCurrentTimeZone = TajoConf.getCurrentTimeZone();
TajoConf.setCurrentTimeZone(TimeZone.getTimeZone("UTC"));
http://git-wip-us.apache.org/repos/asf/tajo/blob/a17bd0f1/tajo-rpc/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java b/tajo-rpc/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
index 0e2db45..72223c1 100644
--- a/tajo-rpc/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
+++ b/tajo-rpc/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
@@ -135,15 +135,21 @@ public class TestAsyncRpc {
assertTrue(future.isDone());
}
- @Test(expected = TimeoutException.class)
+ @Test
public void testCallFutureTimeout() throws Exception {
- EchoMessage echoMessage = EchoMessage.newBuilder()
- .setMessage(MESSAGE).build();
- CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
- stub.deley(null, echoMessage, future);
-
- assertFalse(future.isDone());
- assertEquals(future.get(1, TimeUnit.SECONDS), echoMessage);
+ boolean timeout = false;
+ try {
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+ stub.deley(null, echoMessage, future);
+
+ assertFalse(future.isDone());
+ future.get(1, TimeUnit.SECONDS);
+ } catch (TimeoutException te) {
+ timeout = true;
+ }
+ assertTrue(timeout);
}
@Test