You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by jh...@apache.org on 2013/09/04 18:12:13 UTC
git commit: TAJO-157: The CSVScanner.isSplittable() function does not
work properly. (jinho)
Updated Branches:
refs/heads/master ffdb1c71c -> 68d72cf52
TAJO-157: The CSVScanner.isSplittable() function does not work properly. (jinho)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/68d72cf5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/68d72cf5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/68d72cf5
Branch: refs/heads/master
Commit: 68d72cf529f688bcef37fee38188d4d593315f45
Parents: ffdb1c7
Author: jinossy <ji...@gmail.com>
Authored: Thu Sep 5 01:08:36 2013 +0900
Committer: jinossy <ji...@gmail.com>
Committed: Thu Sep 5 01:08:36 2013 +0900
----------------------------------------------------------------------
CHANGES.txt | 3 +++
.../src/main/java/org/apache/tajo/conf/TajoConf.java | 1 +
.../tajo/engine/planner/PhysicalPlannerImpl.java | 4 ++--
.../apache/tajo/master/YarnTaskRunnerLauncherImpl.java | 7 +++----
.../org/apache/tajo/worker/TajoResourceAllocator.java | 13 +++----------
.../src/main/java/org/apache/tajo/storage/CSVFile.java | 7 +++++--
.../java/org/apache/tajo/storage/StorageManager.java | 2 +-
7 files changed, 18 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/68d72cf5/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 1e8edbb..6acfa57 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -126,6 +126,9 @@ Release 0.2.0 - unreleased
BUG FIXES
+ TAJO-157: The CSVScanner.isSplittable() function does not work properly.
+ (jinho)
+
TAJO-156: BufferUnderflowException occurs during processing the
intermediate file. (jinho)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/68d72cf5/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 9bdda7e..3e9f5e9 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
@@ -106,6 +106,7 @@ public class TajoConf extends YarnConfiguration {
//////////////////////////////////
EXT_SORT_BUFFER("tajo.extsort.buffer", 400000),
BROADCAST_JOIN_THRESHOLD("tajo.join.broadcast.threshold", (long)5 * 1048576),
+ HASH_AGGREGATION_THRESHOLD("tajo.aggregation.hash.threshold", (long)512 * 1048576),
//////////////////////////////////////////
// Distributed Query Execution Parameters
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/68d72cf5/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index 06ac2d6..2d123bd 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -248,9 +248,9 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
} else {
String [] outerLineage = PlannerUtil.getLineage(groupbyNode.getChild());
long estimatedSize = estimateSizeRecursive(ctx, outerLineage);
- final long threshold = 1048576 * 256;
+ final long threshold = conf.getLongVar(TajoConf.ConfVars.HASH_AGGREGATION_THRESHOLD);
- // if the relation size is less than the reshold,
+ // if the relation size is less than the threshold,
// the hash aggregation will be used.
if (estimatedSize <= threshold) {
LOG.info("The planner chooses [Hash Aggregation]");
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/68d72cf5/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
index 472d61d..c4e2468 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
@@ -35,9 +35,7 @@ import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.master.TaskRunnerGroupEvent.EventType;
import org.apache.tajo.master.querymaster.QueryMasterTask;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -81,7 +79,8 @@ public class YarnTaskRunnerLauncherImpl extends AbstractService implements TaskR
executorService.shutdownNow();
Map<ContainerId, ContainerProxy> containers = context.getResourceAllocator().getContainers();
- for(ContainerProxy eachProxy: containers.values()) {
+ List<ContainerProxy> list = new ArrayList<ContainerProxy>(containers.values());
+ for(ContainerProxy eachProxy: list) {
try {
eachProxy.stopContainer();
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/68d72cf5/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
index 1e9d6e1..33f9b0f 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
@@ -112,23 +112,16 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
}
@Override
- public void stop() {
+ public synchronized void stop() {
if(stopped.get()) {
return;
}
stopped.set(true);
executorService.shutdownNow();
- while(!executorService.isTerminated()) {
- LOG.info("====>executorService.isTerminated:" + executorService.isTerminated() + "," +
- executorService.isShutdown());
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- }
- }
Map<ContainerId, ContainerProxy> containers = queryContext.getResourceAllocator().getContainers();
- for(ContainerProxy eachProxy: containers.values()) {
+ List<ContainerProxy> list = new ArrayList<ContainerProxy>(containers.values());
+ for(ContainerProxy eachProxy: list) {
try {
eachProxy.stopContainer();
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/68d72cf5/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
index 9231ead..5e453b5 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
@@ -249,6 +249,9 @@ public class CSVFile {
super(conf, meta, fragment);
factory = new CompressionCodecFactory(conf);
codec = factory.getCodec(fragment.getPath());
+ if (isCompress() && !(codec instanceof SplittableCompressionCodec)) {
+ splittable = false;
+ }
}
private final static int DEFAULT_BUFFER_SIZE = 256 * 1024;
@@ -302,7 +305,6 @@ public class CSVFile {
is = cIn;
} else {
is = new DataInputStream(codec.createInputStream(fis, decompressor));
- splittable = false;
}
} else {
fis.seek(startOffset);
@@ -321,7 +323,7 @@ public class CSVFile {
}
super.init();
- if(LOG.isDebugEnabled()) {
+ if (LOG.isDebugEnabled()) {
LOG.debug("CSVScanner open:" + fragment.getPath() + "," + startOffset + "," + length +
"," + fs.getFileStatus(fragment.getPath()).getLen());
}
@@ -533,6 +535,7 @@ public class CSVFile {
is.close();
} finally {
if (decompressor != null) {
+ decompressor.reset();
CodecPool.returnDecompressor(decompressor);
decompressor = null;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/68d72cf5/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
index db5b5d8..6de58e7 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
@@ -655,7 +655,7 @@ public class StorageManager {
}
}
- LOG.debug("Total # of splits: " + splits.size());
+ LOG.info("Total # of splits: " + splits.size());
return splits;
}