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 2015/03/03 14:05:18 UTC
[1/2] tajo git commit: TAJO-1362: Resolve findbug warnings on Tajo
Core Module
Repository: tajo
Updated Branches:
refs/heads/master f884e59a9 -> 64e47a401
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMasterTask.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMasterTask.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMasterTask.java
index fd52488..0d1924b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMasterTask.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMasterTask.java
@@ -129,6 +129,9 @@ public class QueryMasterTask extends CompositeService {
@Override
public void init(Configuration conf) {
+ if (!(conf instanceof TajoConf)) {
+ throw new IllegalArgumentException("conf should be a TajoConf type.");
+ }
systemConf = (TajoConf)conf;
try {
@@ -377,8 +380,7 @@ public class QueryMasterTask extends CompositeService {
}
private void initStagingDir() throws IOException {
- Path stagingDir = null;
- FileSystem defaultFS = TajoConf.getWarehouseDir(systemConf).getFileSystem(systemConf);
+ Path stagingDir;
try {
@@ -388,14 +390,7 @@ public class QueryMasterTask extends CompositeService {
LOG.info("The staging dir '" + stagingDir + "' is created.");
queryContext.setStagingDir(stagingDir);
} catch (IOException ioe) {
- if (stagingDir != null && defaultFS.exists(stagingDir)) {
- try {
- defaultFS.delete(stagingDir, true);
- LOG.info("The staging directory '" + stagingDir + "' is deleted");
- } catch (Exception e) {
- LOG.warn(e.getMessage());
- }
- }
+ LOG.warn("Creating staging dir has been failed.", ioe);
throw ioe;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
index 7c2d9f4..8e9e343 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
@@ -82,7 +82,6 @@ public class Repartitioner {
public static void scheduleFragmentsForJoinQuery(TaskSchedulerContext schedulerContext, Stage stage)
throws IOException {
- MasterPlan masterPlan = stage.getMasterPlan();
ExecutionBlock execBlock = stage.getBlock();
QueryMasterTask.QueryMasterTaskContext masterContext = stage.getContext();
@@ -231,12 +230,13 @@ public class Repartitioner {
String namePrefix = "";
long maxStats = Long.MIN_VALUE;
int maxStatsScanIdx = -1;
+ StringBuilder nonLeafScanNamesBuilder = new StringBuilder();
for (int i = 0; i < scans.length; i++) {
if (scans[i].getTableDesc().getMeta().getStoreType() == StoreType.RAW) {
// Intermediate data scan
hasNonLeafNode = true;
largeScanIndexList.add(i);
- nonLeafScanNames += namePrefix + scans[i].getCanonicalName();
+ nonLeafScanNamesBuilder.append(namePrefix).append(scans[i].getCanonicalName());
namePrefix = ",";
}
if (execBlock.isBroadcastTable(scans[i].getCanonicalName())) {
@@ -249,18 +249,19 @@ public class Repartitioner {
}
}
}
+ nonLeafScanNames = nonLeafScanNamesBuilder.toString();
if (maxStatsScanIdx == -1) {
maxStatsScanIdx = 0;
}
if (!hasNonLeafNode) {
if (largeScanIndexList.size() > 1) {
- String largeTableNames = "";
+ StringBuilder largeTableNamesBuilder = new StringBuilder();
for (Integer eachId : largeScanIndexList) {
- largeTableNames += scans[eachId].getTableName() + ",";
+ largeTableNamesBuilder.append(scans[eachId].getTableName()).append(',');
}
throw new IOException("Broadcast join with leaf node should have only one large table, " +
- "but " + largeScanIndexList.size() + ", tables=" + largeTableNames);
+ "but " + largeScanIndexList.size() + ", tables=" + largeTableNamesBuilder.toString());
}
int baseScanIdx = largeScanIndexList.isEmpty() ? maxStatsScanIdx : largeScanIndexList.get(0);
LOG.info(String.format("[Distributed Join Strategy] : Broadcast Join, base_table=%s, base_volume=%d",
@@ -494,6 +495,9 @@ public class Repartitioner {
public static List<Fragment> getFragmentsFromPartitionedTable(FileStorageManager sm,
ScanNode scan,
TableDesc table) throws IOException {
+ if (!(scan instanceof PartitionedTableScanNode)) {
+ throw new IllegalArgumentException("scan should be a PartitionedTableScanNode type.");
+ }
List<Fragment> fragments = Lists.newArrayList();
PartitionedTableScanNode partitionsScan = (PartitionedTableScanNode) scan;
fragments.addAll(sm.getSplits(
@@ -696,9 +700,14 @@ public class Repartitioner {
LOG.info(stage.getId() + ", Try to divide " + mergedRange + " into " + determinedTaskNum +
" sub ranges (total units: " + determinedTaskNum + ")");
ranges = partitioner.partition(determinedTaskNum);
- if (ranges == null || ranges.length == 0) {
+ if (ranges == null) {
+ throw new NullPointerException("ranges is null on " + stage.getId() + " stage.");
+ }
+
+ if (ranges.length == 0) {
LOG.warn(stage.getId() + " no range infos.");
}
+
TupleUtil.setMaxRangeIfNull(sortSpecs, sortSchema, totalStat.getColumnStats(), ranges);
if (LOG.isDebugEnabled()) {
if (ranges != null) {
@@ -985,7 +994,7 @@ public class Repartitioner {
String tableName) {
long splitVolume = StorageUnit.MB *
stage.getMasterPlan().getContext().getLong(SessionVars.TABLE_PARTITION_PER_SHUFFLE_SIZE);
- long pageSize = StorageUnit.MB *
+ long pageSize = ((long)StorageUnit.MB) *
stage.getContext().getConf().getIntVar(ConfVars.SHUFFLE_HASH_APPENDER_PAGE_VOLUME); // in bytes
if (pageSize >= splitVolume) {
throw new RuntimeException("tajo.dist-query.table-partition.task-volume-mb should be great than " +
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/querymaster/Stage.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/Stage.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/Stage.java
index efadaa7..4e1f716 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/Stage.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/Stage.java
@@ -865,7 +865,6 @@ public class Stage implements EventHandler<StageEvent> {
* @return
*/
public static int calculateShuffleOutputNum(Stage stage, DataChannel channel) {
- TajoConf conf = stage.context.getConf();
MasterPlan masterPlan = stage.getMasterPlan();
ExecutionBlock parent = masterPlan.getParent(stage.getBlock());
@@ -1156,6 +1155,9 @@ public class Stage implements EventHandler<StageEvent> {
@Override
public void transition(Stage stage, StageEvent event) {
+ if (!(event instanceof StageContainerAllocationEvent)) {
+ throw new IllegalArgumentException("event should be a StageContainerAllocationEvent type.");
+ }
try {
StageContainerAllocationEvent allocationEvent =
(StageContainerAllocationEvent) event;
@@ -1191,6 +1193,9 @@ public class Stage implements EventHandler<StageEvent> {
private static class AllocatedContainersCancelTransition implements SingleArcTransition<Stage, StageEvent> {
@Override
public void transition(Stage stage, StageEvent event) {
+ if (!(event instanceof StageContainerAllocationEvent)) {
+ throw new IllegalArgumentException("event should be a StageContainerAllocationEvent type.");
+ }
try {
StageContainerAllocationEvent allocationEvent =
(StageContainerAllocationEvent) event;
@@ -1213,6 +1218,9 @@ public class Stage implements EventHandler<StageEvent> {
@Override
public void transition(Stage stage,
StageEvent event) {
+ if (!(event instanceof StageTaskEvent)) {
+ throw new IllegalArgumentException("event should be a StageTaskEvent type.");
+ }
StageTaskEvent taskEvent = (StageTaskEvent) event;
Task task = stage.getTask(taskEvent.getTaskId());
@@ -1418,6 +1426,9 @@ public class Stage implements EventHandler<StageEvent> {
private static class DiagnosticsUpdateTransition implements SingleArcTransition<Stage, StageEvent> {
@Override
public void transition(Stage stage, StageEvent event) {
+ if (!(event instanceof StageDiagnosticsUpdateEvent)) {
+ throw new IllegalArgumentException("event should be a StageDiagnosticsUpdateEvent type.");
+ }
stage.addDiagnostic(((StageDiagnosticsUpdateEvent) event).getDiagnosticUpdate());
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/querymaster/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/Task.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/Task.java
index ad01b62..92f4b20 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/Task.java
@@ -601,6 +601,9 @@ public class Task implements EventHandler<TaskEvent> {
@Override
public void transition(Task task,
TaskEvent event) {
+ if (!(event instanceof TaskTAttemptEvent)) {
+ throw new IllegalArgumentException("event should be a TaskTAttemptEvent type.");
+ }
TaskTAttemptEvent attemptEvent = (TaskTAttemptEvent) event;
TaskAttempt attempt = task.attempts.get(attemptEvent.getTaskAttemptId());
@@ -618,6 +621,9 @@ public class Task implements EventHandler<TaskEvent> {
@Override
public void transition(Task task,
TaskEvent event) {
+ if (!(event instanceof TaskTAttemptEvent)) {
+ throw new IllegalArgumentException("event should be a TaskTAttemptEvent type.");
+ }
TaskTAttemptEvent attemptEvent = (TaskTAttemptEvent) event;
TaskAttempt attempt = task.attempts.get(attemptEvent.getTaskAttemptId());
task.launchTime = System.currentTimeMillis();
@@ -629,6 +635,9 @@ public class Task implements EventHandler<TaskEvent> {
private static class AttemptFailedTransition implements SingleArcTransition<Task, TaskEvent> {
@Override
public void transition(Task task, TaskEvent event) {
+ if (!(event instanceof TaskTAttemptEvent)) {
+ throw new IllegalArgumentException("event should be a TaskTAttemptEvent type.");
+ }
TaskTAttemptEvent attemptEvent = (TaskTAttemptEvent) event;
LOG.info("=============================================================");
LOG.info(">>> Task Failed: " + attemptEvent.getTaskAttemptId() + " <<<");
@@ -646,6 +655,9 @@ public class Task implements EventHandler<TaskEvent> {
@Override
public TaskState transition(Task task, TaskEvent taskEvent) {
+ if (!(taskEvent instanceof TaskTAttemptEvent)) {
+ throw new IllegalArgumentException("taskEvent should be a TaskTAttemptEvent type.");
+ }
TaskTAttemptEvent attemptEvent = (TaskTAttemptEvent) taskEvent;
task.failedAttempts++;
task.finishedAttempts++;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/querymaster/TaskAttempt.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/TaskAttempt.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/TaskAttempt.java
index 86c49b4..c1b9273 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/TaskAttempt.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/TaskAttempt.java
@@ -307,6 +307,9 @@ public class TaskAttempt implements EventHandler<TaskAttemptEvent> {
@Override
public void transition(TaskAttempt taskAttempt,
TaskAttemptEvent event) {
+ if (!(event instanceof TaskAttemptAssignedEvent)) {
+ throw new IllegalArgumentException("event should be a TaskAttemptAssignedEvent type.");
+ }
TaskAttemptAssignedEvent castEvent = (TaskAttemptAssignedEvent) event;
taskAttempt.containerId = castEvent.getContainerId();
taskAttempt.workerConnectionInfo = castEvent.getWorkerConnectionInfo();
@@ -333,6 +336,9 @@ public class TaskAttempt implements EventHandler<TaskAttemptEvent> {
@Override
public TaskAttemptState transition(TaskAttempt taskAttempt,
TaskAttemptEvent event) {
+ if (!(event instanceof TaskAttemptStatusUpdateEvent)) {
+ throw new IllegalArgumentException("event should be a TaskAttemptStatusUpdateEvent type.");
+ }
TaskAttemptStatusUpdateEvent updateEvent = (TaskAttemptStatusUpdateEvent) event;
taskAttempt.progress = updateEvent.getStatus().getProgress();
@@ -371,6 +377,9 @@ public class TaskAttempt implements EventHandler<TaskAttemptEvent> {
@Override
public void transition(TaskAttempt taskAttempt,
TaskAttemptEvent event) {
+ if (!(event instanceof TaskCompletionEvent)) {
+ throw new IllegalArgumentException("event should be a TaskCompletionEvent type.");
+ }
TaskCompletionReport report = ((TaskCompletionEvent)event).getReport();
try {
@@ -395,6 +404,9 @@ public class TaskAttempt implements EventHandler<TaskAttemptEvent> {
private static class FailedTransition implements SingleArcTransition<TaskAttempt, TaskAttemptEvent>{
@Override
public void transition(TaskAttempt taskAttempt, TaskAttemptEvent event) {
+ if (!(event instanceof TaskFatalErrorEvent)) {
+ throw new IllegalArgumentException("event should be a TaskFatalErrorEvent type.");
+ }
TaskFatalErrorEvent errorEvent = (TaskFatalErrorEvent) event;
taskAttempt.eventHandler.handle(new TaskTAttemptEvent(taskAttempt.getId(), TaskEventType.T_ATTEMPT_FAILED));
taskAttempt.addDiagnosticInfo(errorEvent.errorMessage());
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/IndexUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/IndexUtil.java b/tajo-core/src/main/java/org/apache/tajo/util/IndexUtil.java
index 3147bb6..23d245b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/IndexUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/IndexUtil.java
@@ -143,6 +143,9 @@ public class IndexUtil {
&& binaryEval.getRightExpr().getType() == EvalType.CONST) {
nodeList.add(binaryEval);
}
+ break;
+ default:
+ break;
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java b/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
index 13f4dcc..875d12b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
@@ -97,6 +97,9 @@ public class JSPUtil {
}
public static String getTajoMasterHttpAddr(Configuration config) {
+ if (!(config instanceof TajoConf)) {
+ throw new IllegalArgumentException("config should be a TajoConf type.");
+ }
try {
TajoConf conf = (TajoConf) config;
String [] masterAddr = conf.getVar(ConfVars.TAJO_MASTER_UMBILICAL_RPC_ADDRESS).split(":");
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java
index 868dfcd..6ba74d5 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java
@@ -127,6 +127,8 @@ public class HistoryCleaner extends Thread {
}
}
}
+ } catch (RuntimeException e) {
+ LOG.error(e.getMessage(), e);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java
index 3fea3ef..b06c7e8 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java
@@ -84,6 +84,9 @@ public class HistoryWriter extends AbstractService {
@Override
public void serviceInit(Configuration conf) throws Exception {
+ if (!(conf instanceof TajoConf)) {
+ throw new IllegalArgumentException("conf should be a TajoConf type.");
+ }
tajoConf = (TajoConf)conf;
historyParentPath = tajoConf.getQueryHistoryDir(tajoConf);
taskHistoryParentPath = tajoConf.getTaskHistoryDir(tajoConf);
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java
index 9e895b8..386fb79 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java
@@ -21,9 +21,7 @@ package org.apache.tajo.util.metrics.reporter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
+import java.io.*;
public class MetricsFileScheduledReporter extends MetricsStreamScheduledReporter {
private static final Log LOG = LogFactory.getLog(MetricsFileScheduledReporter.class);
@@ -40,6 +38,7 @@ public class MetricsFileScheduledReporter extends MetricsStreamScheduledReporter
LOG.warn("No " + metricsPropertyKey + "filename property in tajo-metrics.properties");
return;
}
+
try {
File file = new File(fileName);
File parentFile = file.getParentFile();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java
index 4fbefd7..7f33792 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java
@@ -36,7 +36,6 @@ public abstract class MetricsStreamScheduledReporter extends TajoMetricsSchedule
protected Locale locale;
protected Clock clock;
protected TimeZone timeZone;
- protected MetricFilter filter;
protected DateFormat dateFormat;
private final byte[] NEW_LINE = "\n".getBytes();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java
index a32a913..a7e0559 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java
@@ -48,10 +48,12 @@ public abstract class TajoMetricsReporter {
groupName = keyTokens[0] + "." + keyTokens[1];
itemName = "";
String prefix = "";
+ StringBuilder itemNameBuilder = new StringBuilder();
for (int i = 2; i < keyTokens.length; i++) {
- itemName += prefix + keyTokens[i];
+ itemNameBuilder.append(prefix).append(keyTokens[i]);
prefix = ".";
}
+ itemName = itemNameBuilder.toString();
} else {
groupName = "";
itemName = key;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsScheduledReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsScheduledReporter.java b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsScheduledReporter.java
index f11d520..7e0ec4a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsScheduledReporter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsScheduledReporter.java
@@ -142,11 +142,7 @@ public abstract class TajoMetricsScheduledReporter extends TajoMetricsReporter i
try {
report();
} catch (Exception e) {
- if(LOG.isDebugEnabled()) {
- LOG.warn("Metric report error:" + e.getMessage(), e);
- } else {
- LOG.warn("Metric report error:" + e.getMessage(), e);
- }
+ LOG.warn("Metric report error:" + e.getMessage(), e);
}
}
}, period, period, unit);
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java b/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
index d8bb8f1..e9b6230 100644
--- a/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
+++ b/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
@@ -28,6 +28,7 @@ import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
+import java.io.NotSerializableException;
import java.io.OutputStream;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
@@ -62,17 +63,26 @@ import java.util.concurrent.atomic.AtomicInteger;
public class QueryExecutorServlet extends HttpServlet {
private static final Log LOG = LogFactory.getLog(QueryExecutorServlet.class);
+ private static final long serialVersionUID = -1517586415463171579L;
- ObjectMapper om = new ObjectMapper();
+ transient ObjectMapper om = new ObjectMapper();
//queryRunnerId -> QueryRunner
//TODO We must handle the session.
- private final Map<String, QueryRunner> queryRunners = new HashMap<String, QueryRunner>();
+ private transient final Map<String, QueryRunner> queryRunners = new HashMap<String, QueryRunner>();
- private TajoConf tajoConf;
- private TajoClient tajoClient;
+ private transient TajoConf tajoConf;
+ private transient TajoClient tajoClient;
- private ExecutorService queryRunnerExecutor = Executors.newFixedThreadPool(5);
+ private transient ExecutorService queryRunnerExecutor = Executors.newFixedThreadPool(5);
+
+ private void writeObject(java.io.ObjectOutputStream stream) throws java.io.IOException {
+ throw new NotSerializableException( getClass().getName() );
+ }
+
+ private void readObject(java.io.ObjectInputStream stream) throws java.io.IOException, ClassNotFoundException {
+ throw new NotSerializableException( getClass().getName() );
+ }
@Override
public void init(ServletConfig config) throws ServletException {
@@ -135,10 +145,11 @@ public class QueryExecutorServlet extends HttpServlet {
if(!queryRunners.containsKey(queryRunnerId)) {
break;
}
- try {
- Thread.sleep(100);
- } catch (InterruptedException e) {
- }
+ }
+
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
}
}
String database = request.getParameter("database");
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/webapp/StaticHttpServer.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/webapp/StaticHttpServer.java b/tajo-core/src/main/java/org/apache/tajo/webapp/StaticHttpServer.java
index 09426e0..6008aae 100644
--- a/tajo-core/src/main/java/org/apache/tajo/webapp/StaticHttpServer.java
+++ b/tajo-core/src/main/java/org/apache/tajo/webapp/StaticHttpServer.java
@@ -31,6 +31,8 @@ import java.net.Inet4Address;
public class StaticHttpServer extends HttpServer {
private static StaticHttpServer instance = null;
+ private static final Object lockObjectForStaticHttpServer = new Object();
+
private StaticHttpServer(Object containerObject , String name, String bindAddress, int port,
boolean findPort, Connector connector, Configuration conf,
String[] pathSpecs) throws IOException {
@@ -52,13 +54,17 @@ public class StaticHttpServer extends HttpServer {
addr = Inet4Address.getLocalHost().getHostName();
}
}
-
- instance = new StaticHttpServer(containerObject, name, addr, port,
- findPort, connector, conf, pathSpecs);
- instance.setAttribute("tajo.info.server.object", containerObject);
- instance.setAttribute("tajo.info.server.addr", addr);
- instance.setAttribute("tajo.info.server.conf", conf);
- instance.setAttribute("tajo.info.server.starttime", System.currentTimeMillis());
+
+ synchronized (lockObjectForStaticHttpServer) {
+ if (instance == null) {
+ instance = new StaticHttpServer(containerObject, name, addr, port,
+ findPort, connector, conf, pathSpecs);
+ instance.setAttribute("tajo.info.server.object", containerObject);
+ instance.setAttribute("tajo.info.server.addr", addr);
+ instance.setAttribute("tajo.info.server.conf", conf);
+ instance.setAttribute("tajo.info.server.starttime", System.currentTimeMillis());
+ }
+ }
}
return instance;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockSharedResource.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockSharedResource.java b/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockSharedResource.java
index 4b5a203..67114a3 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockSharedResource.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockSharedResource.java
@@ -37,7 +37,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class ExecutionBlockSharedResource {
private static Log LOG = LogFactory.getLog(ExecutionBlockSharedResource.class);
private AtomicBoolean initializing = new AtomicBoolean(false);
- private volatile Boolean resourceInitSuccess = new Boolean(false);
+ private volatile Boolean resourceInitSuccess = Boolean.valueOf(false);
// Query
private QueryContext context;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
index c6a06f0..827c860 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
@@ -106,6 +106,9 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
@Override
public void init(Configuration conf) {
+ if (!(conf instanceof TajoConf)) {
+ throw new IllegalArgumentException("conf should be a TajoConf type.");
+ }
tajoConf = (TajoConf)conf;
queryTaskContext.getDispatcher().register(TaskRunnerGroupEvent.EventType.class, new TajoTaskRunnerLauncher());
@@ -147,6 +150,9 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
@Override
public void handle(TaskRunnerGroupEvent event) {
if (event.getType() == TaskRunnerGroupEvent.EventType.CONTAINER_REMOTE_LAUNCH) {
+ if (!(event instanceof LaunchTaskRunnersEvent)) {
+ throw new IllegalArgumentException("event should be a LaunchTaskRunnersEvent type.");
+ }
LaunchTaskRunnersEvent launchEvent = (LaunchTaskRunnersEvent) event;
launchTaskRunners(launchEvent);
} else if (event.getType() == TaskRunnerGroupEvent.EventType.CONTAINER_REMOTE_CLEANUP) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
index 7f73916..7e2a233 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -186,6 +186,9 @@ public class TajoWorker extends CompositeService {
@Override
public void serviceInit(Configuration conf) throws Exception {
+ if (!(conf instanceof TajoConf)) {
+ throw new IllegalArgumentException("conf should be a TajoConf type.");
+ }
Runtime.getRuntime().addShutdownHook(new Thread(new ShutdownHook()));
this.systemConf = (TajoConf)conf;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
index 23efffa..cf50767 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
@@ -113,6 +113,9 @@ public class TaskRunner extends AbstractService {
@Override
public void init(Configuration conf) {
+ if (!(conf instanceof TajoConf)) {
+ throw new IllegalArgumentException("conf should be a TajoConf Type.");
+ }
this.systemConf = (TajoConf)conf;
try {
@@ -143,15 +146,17 @@ public class TaskRunner extends AbstractService {
this.finishTime = System.currentTimeMillis();
this.history.setFinishTime(finishTime);
// If this flag become true, taskLauncher will be terminated.
- this.stopped = true;
-
- fetchLauncher.shutdown();
- fetchLauncher = null;
LOG.info("Stop TaskRunner: " + getId());
synchronized (this) {
+ this.stopped = true;
+
+ fetchLauncher.shutdown();
+ fetchLauncher = null;
+
notifyAll();
}
+
super.stop();
this.history.setState(getServiceState());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
index 57ae566..570bd38 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
@@ -230,7 +230,7 @@ public class TaskRunnerManager extends CompositeService implements EventHandler<
break;
}
try {
- long expireTime = System.currentTimeMillis() - expireIntervalTime * 60 * 1000;
+ long expireTime = System.currentTimeMillis() - expireIntervalTime * 60 * 1000l;
cleanExpiredFinishedQueryMasterTask(expireTime);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java b/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
index 870e9a0..462f95d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
@@ -111,7 +111,6 @@ public class WorkerHeartbeatService extends AbstractService {
int workerCpuCoreNum;
boolean dedicatedResource = systemConf.getBoolVar(TajoConf.ConfVars.WORKER_RESOURCE_DEDICATED);
- int workerCpuCoreSlots = Runtime.getRuntime().availableProcessors();
try {
diskDeviceInfos = DiskUtil.getDiskDeviceInfos();
[2/2] tajo git commit: TAJO-1362: Resolve findbug warnings on Tajo
Core Module
Posted by ji...@apache.org.
TAJO-1362: Resolve findbug warnings on Tajo Core Module
Closes #390
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/64e47a40
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/64e47a40
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/64e47a40
Branch: refs/heads/master
Commit: 64e47a401fca83bef46567631940ae206ebfb280
Parents: f884e59
Author: Jihun Kang <ji...@apache.org>
Authored: Tue Mar 3 22:04:36 2015 +0900
Committer: Jihun Kang <ji...@apache.org>
Committed: Tue Mar 3 22:04:36 2015 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../tajo/engine/codegen/EvalCodeGenerator.java | 6 ++-
.../engine/codegen/TajoGeneratorAdapter.java | 26 +++++++++--
.../tajo/engine/function/builtin/AvgDouble.java | 2 +-
.../tajo/engine/function/builtin/AvgLong.java | 2 +-
.../tajo/engine/function/builtin/CountRows.java | 2 +-
.../function/builtin/CountValueDistinct.java | 2 +-
.../tajo/engine/function/builtin/LastValue.java | 2 +-
.../tajo/engine/function/builtin/Lead.java | 2 +-
.../tajo/engine/function/builtin/Max.java | 2 +-
.../tajo/engine/function/builtin/Min.java | 2 +-
.../tajo/engine/function/builtin/SumDouble.java | 2 +-
.../function/builtin/SumDoubleDistinct.java | 2 +-
.../function/builtin/SumFloatDistinct.java | 2 +-
.../engine/function/builtin/SumIntDistinct.java | 2 +-
.../tajo/engine/function/builtin/SumLong.java | 2 +-
.../function/builtin/SumLongDistinct.java | 2 +-
.../function/datetime/DatePartFromDate.java | 28 ++++++------
.../function/datetime/DatePartFromTime.java | 18 +++-----
.../datetime/DatePartFromTimestamp.java | 40 ++++++++---------
.../datetime/DateTimePartFromUnixTimestamp.java | 10 ++---
.../engine/function/string/RegexpReplace.java | 2 -
.../tajo/engine/function/window/FirstValue.java | 2 +-
.../apache/tajo/engine/function/window/Lag.java | 2 +-
.../tajo/engine/function/window/Rank.java | 2 +-
.../tajo/engine/function/window/RowNumber.java | 2 +-
.../engine/planner/PhysicalPlannerImpl.java | 24 +++++-----
.../planner/global/ExecutionBlockCursor.java | 8 ++++
.../engine/planner/global/GlobalPlanner.java | 6 +--
.../global/builder/DistinctGroupbyBuilder.java | 8 +---
.../planner/physical/AggregationExec.java | 7 ---
.../engine/planner/physical/BNLJoinExec.java | 2 +-
.../planner/physical/ColPartitionStoreExec.java | 6 +++
.../DistinctGroupbyFirstAggregationExec.java | 4 --
.../DistinctGroupbyHashAggregationExec.java | 8 ----
.../DistinctGroupbySortAggregationExec.java | 3 --
.../DistinctGroupbyThirdAggregationExec.java | 6 +--
.../planner/physical/ExternalSortExec.java | 2 +-
.../engine/planner/physical/HashJoinExec.java | 2 -
.../physical/HashShuffleFileWriteExec.java | 5 ++-
.../physical/RightOuterMergeJoinExec.java | 1 -
.../engine/planner/physical/SeqScanExec.java | 4 +-
.../engine/planner/physical/WindowAggExec.java | 2 -
.../apache/tajo/engine/utils/TupleCacheKey.java | 3 ++
.../tajo/engine/utils/TupleCacheScanner.java | 2 +-
.../org/apache/tajo/ha/HdfsServiceTracker.java | 7 ++-
.../java/org/apache/tajo/master/QueryInfo.java | 47 ++++++++++++++++++++
.../apache/tajo/master/TajoContainerProxy.java | 2 +-
.../java/org/apache/tajo/master/TajoMaster.java | 3 ++
.../NonForwardQueryResultSystemScanner.java | 9 +++-
.../tajo/master/rm/TajoResourceTracker.java | 4 +-
.../tajo/master/rm/TajoWorkerContainer.java | 22 +++++++++
.../java/org/apache/tajo/master/rm/Worker.java | 37 +++++++++++++++
.../java/org/apache/tajo/querymaster/Query.java | 17 +++++--
.../apache/tajo/querymaster/QueryMaster.java | 8 +++-
.../tajo/querymaster/QueryMasterTask.java | 15 +++----
.../apache/tajo/querymaster/Repartitioner.java | 23 +++++++---
.../java/org/apache/tajo/querymaster/Stage.java | 13 +++++-
.../java/org/apache/tajo/querymaster/Task.java | 12 +++++
.../apache/tajo/querymaster/TaskAttempt.java | 12 +++++
.../java/org/apache/tajo/util/IndexUtil.java | 3 ++
.../main/java/org/apache/tajo/util/JSPUtil.java | 3 ++
.../tajo/util/history/HistoryCleaner.java | 2 +
.../apache/tajo/util/history/HistoryWriter.java | 3 ++
.../reporter/MetricsFileScheduledReporter.java | 5 +--
.../MetricsStreamScheduledReporter.java | 1 -
.../metrics/reporter/TajoMetricsReporter.java | 4 +-
.../reporter/TajoMetricsScheduledReporter.java | 6 +--
.../tajo/webapp/QueryExecutorServlet.java | 29 ++++++++----
.../apache/tajo/webapp/StaticHttpServer.java | 20 ++++++---
.../worker/ExecutionBlockSharedResource.java | 2 +-
.../tajo/worker/TajoResourceAllocator.java | 6 +++
.../java/org/apache/tajo/worker/TajoWorker.java | 3 ++
.../java/org/apache/tajo/worker/TaskRunner.java | 13 ++++--
.../apache/tajo/worker/TaskRunnerManager.java | 2 +-
.../tajo/worker/WorkerHeartbeatService.java | 1 -
76 files changed, 412 insertions(+), 193 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 716dc59..e8c8b18 100644
--- a/CHANGES
+++ b/CHANGES
@@ -10,6 +10,8 @@ Release 0.11.0 - unreleased
BUG FIXES
+ TAJO-1362: Resolve findbug warnings on Tajo Core Module. (jihun)
+
TAJO-1366: The timestamp type conversion occasionally leads to wrong
results. (jihoon)
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenerator.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenerator.java b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenerator.java
index 69252c4..1cb3755 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenerator.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenerator.java
@@ -73,11 +73,15 @@ public class EvalCodeGenerator extends SimpleEvalNodeVisitor<EvalCodeGenContext>
this.classLoader = classLoader;
}
+ protected synchronized static int getAndIncrementClassSeq() {
+ return classSeq++;
+ }
+
public EvalNode compile(Schema schema, EvalNode expr) throws CompilationError {
ClassWriter classWriter = new ClassWriter(ClassWriter.COMPUTE_MAXS);
- String className = EvalCodeGenerator.class.getPackage().getName() + ".CompiledEval" + classSeq++;
+ String className = EvalCodeGenerator.class.getPackage().getName() + ".CompiledEval" + getAndIncrementClassSeq();
EvalCodeGenContext context = new EvalCodeGenContext(TajoGeneratorAdapter.getInternalName(className),
schema, classWriter, expr);
visit(context, expr, new Stack<EvalNode>());
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/codegen/TajoGeneratorAdapter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/TajoGeneratorAdapter.java b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/TajoGeneratorAdapter.java
index 9cc1dc3..21cad7e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/TajoGeneratorAdapter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/TajoGeneratorAdapter.java
@@ -184,7 +184,7 @@ class TajoGeneratorAdapter {
} else if (value >= Short.MIN_VALUE && value <= Short.MAX_VALUE) {
methodvisitor.visitIntInsn(Opcodes.SIPUSH, value);
} else {
- methodvisitor.visitLdcInsn(new Integer(value));
+ methodvisitor.visitLdcInsn(Integer.valueOf(value));
}
}
@@ -192,7 +192,7 @@ class TajoGeneratorAdapter {
if (value == 0L || value == 1L) {
methodvisitor.visitInsn(Opcodes.LCONST_0 + (int) value);
} else {
- methodvisitor.visitLdcInsn(new Long(value));
+ methodvisitor.visitLdcInsn(Long.valueOf(value));
}
}
@@ -201,7 +201,7 @@ class TajoGeneratorAdapter {
if (bits == 0L || bits == 0x3f800000 || bits == 0x40000000) { // 0..2
methodvisitor.visitInsn(Opcodes.FCONST_0 + (int) value);
} else {
- methodvisitor.visitLdcInsn(new Float(value));
+ methodvisitor.visitLdcInsn(Float.valueOf(value));
}
}
@@ -1009,5 +1009,25 @@ class TajoGeneratorAdapter {
public int compareTo(SwitchCase o) {
return index - o.index;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ SwitchCase that = (SwitchCase) o;
+
+ if (index != that.index) return false;
+ if (thanResult != null ? !thanResult.equals(that.thanResult) : that.thanResult != null) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = index;
+ result = 31 * result + (thanResult != null ? thanResult.hashCode() : 0);
+ return result;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgDouble.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgDouble.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgDouble.java
index a69beca..3ccb02d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgDouble.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgDouble.java
@@ -105,7 +105,7 @@ public class AvgDouble extends AggFunction<Datum> {
return DatumFactory.createFloat8(avgCtx.sum / avgCtx.count);
}
- protected class AvgContext implements FunctionContext {
+ protected static class AvgContext implements FunctionContext {
double sum = 0.0;
long count = 0;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgLong.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgLong.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgLong.java
index 417ecb7..88f8486 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgLong.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/AvgLong.java
@@ -103,7 +103,7 @@ public class AvgLong extends AggFunction<Datum> {
return DatumFactory.createFloat8((double) avgCtx.sum / avgCtx.count);
}
- protected class AvgContext implements FunctionContext {
+ protected static class AvgContext implements FunctionContext {
long sum = 0;
long count = 0;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountRows.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountRows.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountRows.java
index e301c1b..5e41dc9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountRows.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountRows.java
@@ -78,7 +78,7 @@ public class CountRows extends AggFunction<Datum> {
return DatumFactory.createInt8(((CountRowContext) ctx).count);
}
- protected class CountRowContext implements FunctionContext {
+ protected static class CountRowContext implements FunctionContext {
long count;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountValueDistinct.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountValueDistinct.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountValueDistinct.java
index 1ffb51b..675193f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountValueDistinct.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/CountValueDistinct.java
@@ -78,7 +78,7 @@ public final class CountValueDistinct extends CountRows {
return new CountDistinctValueContext();
}
- private class CountDistinctValueContext implements FunctionContext {
+ private static class CountDistinctValueContext implements FunctionContext {
long count = 0;
Datum latest = null;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/LastValue.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/LastValue.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/LastValue.java
index db4f79f..16edb17 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/LastValue.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/LastValue.java
@@ -61,7 +61,7 @@ public abstract class LastValue extends AggFunction<Datum> {
}
}
- private class LastValueContext implements FunctionContext {
+ private static class LastValueContext implements FunctionContext {
Datum last = null;
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Lead.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Lead.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Lead.java
index dc135db..2e95eb7 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Lead.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Lead.java
@@ -84,7 +84,7 @@ public abstract class Lead extends AggFunction<Datum> {
}
}
- private class LeadContext implements FunctionContext {
+ private static class LeadContext implements FunctionContext {
LinkedList<Datum> leadBuffer = new LinkedList<Datum>();
int leadNum = -1;
Datum defaultDatum = null;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Max.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Max.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Max.java
index f851994..6931ffc 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Max.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Max.java
@@ -70,7 +70,7 @@ public abstract class Max extends AggFunction<Datum> {
}
}
- private class MaxContext implements FunctionContext {
+ private static class MaxContext implements FunctionContext {
Datum max = null;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Min.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Min.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Min.java
index 2695b6a..9a2e3ad 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Min.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/Min.java
@@ -70,7 +70,7 @@ public abstract class Min extends AggFunction<Datum> {
}
}
- private class MinContext implements FunctionContext {
+ private static class MinContext implements FunctionContext {
Datum min = null;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDouble.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDouble.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDouble.java
index 2f42272..ca33b6f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDouble.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDouble.java
@@ -95,7 +95,7 @@ public class SumDouble extends AggFunction<Datum> {
}
}
- protected class SumContext implements FunctionContext {
+ protected static class SumContext implements FunctionContext {
boolean hasNonNull = false;
double sum = 0.0;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDoubleDistinct.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDoubleDistinct.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDoubleDistinct.java
index e2a2e0d..76c79f2 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDoubleDistinct.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumDoubleDistinct.java
@@ -87,7 +87,7 @@ public class SumDoubleDistinct extends AggFunction<Datum> {
return DatumFactory.createFloat8(((SumContext) ctx).sum);
}
- private class SumContext implements FunctionContext {
+ private static class SumContext implements FunctionContext {
double sum;
Datum latest;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumFloatDistinct.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumFloatDistinct.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumFloatDistinct.java
index a32315d..a1855b4 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumFloatDistinct.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumFloatDistinct.java
@@ -87,7 +87,7 @@ public class SumFloatDistinct extends AggFunction<Datum> {
return DatumFactory.createFloat8(((SumContext) ctx).sum);
}
- private class SumContext implements FunctionContext {
+ private static class SumContext implements FunctionContext {
double sum;
Datum latest;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumIntDistinct.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumIntDistinct.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumIntDistinct.java
index eb3237b..77eb7a6 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumIntDistinct.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumIntDistinct.java
@@ -87,7 +87,7 @@ public class SumIntDistinct extends AggFunction<Datum> {
return DatumFactory.createInt8(((SumContext) ctx).sum);
}
- private class SumContext implements FunctionContext {
+ private static class SumContext implements FunctionContext {
long sum;
Datum latest;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLong.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLong.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLong.java
index 55484d7..6503a83 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLong.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLong.java
@@ -95,7 +95,7 @@ public class SumLong extends AggFunction<Datum> {
}
}
- protected class SumContext implements FunctionContext {
+ protected static class SumContext implements FunctionContext {
boolean hasNonNull;
long sum = 0;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLongDistinct.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLongDistinct.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLongDistinct.java
index 368ede0..f22a483 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLongDistinct.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/builtin/SumLongDistinct.java
@@ -87,7 +87,7 @@ public class SumLongDistinct extends AggFunction<Datum> {
return DatumFactory.createInt8(((SumContext) ctx).sum);
}
- private class SumContext implements FunctionContext {
+ private static class SumContext implements FunctionContext {
long sum;
Datum latest;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromDate.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromDate.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromDate.java
index 31ecfd9..069a7ee 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromDate.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromDate.java
@@ -100,91 +100,91 @@ public class DatePartFromDate extends GeneralFunction {
public Datum extract(DateDatum date);
}
- private class CenturyExtractorFromDate implements DatePartExtractorFromDate {
+ private static class CenturyExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getCenturyOfEra());
}
}
- private class DayExtractorFromDate implements DatePartExtractorFromDate {
+ private static class DayExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getDayOfMonth());
}
}
- private class DecadeExtractorFromDate implements DatePartExtractorFromDate {
+ private static class DecadeExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
- return DatumFactory.createFloat8((double) (date.getYear() / 10));
+ return DatumFactory.createFloat8((date.getYear() / 10.0d));
}
}
- private class DowExtractorFromDate implements DatePartExtractorFromDate {
+ private static class DowExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getDayOfWeek());
}
}
- private class DoyExtractorFromDate implements DatePartExtractorFromDate {
+ private static class DoyExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getDayOfYear());
}
}
- private class ISODowExtractorFromDate implements DatePartExtractorFromDate {
+ private static class ISODowExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getISODayOfWeek());
}
}
- private class ISOYearExtractorFromDate implements DatePartExtractorFromDate {
+ private static class ISOYearExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getWeekyear());
}
}
- private class MillenniumExtractorFromDate implements DatePartExtractorFromDate {
+ private static class MillenniumExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) (((date.getYear() - 1) / 1000) + 1));
}
}
- private class MonthExtractorFromDate implements DatePartExtractorFromDate {
+ private static class MonthExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getMonthOfYear());
}
}
- private class QuarterExtractorFromDate implements DatePartExtractorFromDate {
+ private static class QuarterExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) (((date.getMonthOfYear() - 1) / 3) + 1));
}
}
- private class WeekExtractorFromDate implements DatePartExtractorFromDate {
+ private static class WeekExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getWeekOfYear());
}
}
- private class YearExtractorFromDate implements DatePartExtractorFromDate {
+ private static class YearExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return DatumFactory.createFloat8((double) date.getYear());
}
}
- private class NullExtractorFromDate implements DatePartExtractorFromDate {
+ private static class NullExtractorFromDate implements DatePartExtractorFromDate {
@Override
public Datum extract(DateDatum date) {
return NullDatum.get();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTime.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTime.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTime.java
index 85c76a2..8100379 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTime.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTime.java
@@ -95,12 +95,6 @@ public class DatePartFromTime extends GeneralFunction {
extractor = new MinuteExtractorFromTime();
} else if (extractType.equals("second")) {
extractor = new SecondExtractorFromTime();
- } else if (extractType.equals("timezone")) {
- extractor = new NullExtractorFromTime();
- } else if (extractType.equals("timezone_hour")) {
- extractor = new NullExtractorFromTime();
- } else if (extractType.equals("timezone_minute")) {
- extractor = new NullExtractorFromTime();
} else {
extractor = new NullExtractorFromTime();
}
@@ -115,35 +109,35 @@ public class DatePartFromTime extends GeneralFunction {
public Datum extract(TimeMeta tm);
}
- private class HourExtractorFromTime implements DatePartExtractorFromTime {
+ private static class HourExtractorFromTime implements DatePartExtractorFromTime {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.hours);
}
}
- private class MicrosecondsExtractorFromTime implements DatePartExtractorFromTime {
+ private static class MicrosecondsExtractorFromTime implements DatePartExtractorFromTime {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) (tm.secs * 1000000 + tm.fsecs));
}
}
- private class MillisecondsExtractorFromTime implements DatePartExtractorFromTime {
+ private static class MillisecondsExtractorFromTime implements DatePartExtractorFromTime {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) (tm.secs * 1000 + tm.fsecs / 1000.0));
}
}
- private class MinuteExtractorFromTime implements DatePartExtractorFromTime {
+ private static class MinuteExtractorFromTime implements DatePartExtractorFromTime {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.minutes);
}
}
- private class SecondExtractorFromTime implements DatePartExtractorFromTime {
+ private static class SecondExtractorFromTime implements DatePartExtractorFromTime {
@Override
public Datum extract(TimeMeta tm) {
if (tm.fsecs != 0) {
@@ -154,7 +148,7 @@ public class DatePartFromTime extends GeneralFunction {
}
}
- private class NullExtractorFromTime implements DatePartExtractorFromTime {
+ private static class NullExtractorFromTime implements DatePartExtractorFromTime {
@Override
public Datum extract(TimeMeta tm) {
return NullDatum.get();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTimestamp.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTimestamp.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTimestamp.java
index 57b3900..a115d00 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTimestamp.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DatePartFromTimestamp.java
@@ -138,28 +138,28 @@ public class DatePartFromTimestamp extends GeneralFunction {
public Datum extract(TimeMeta tm);
}
- private class CenturyExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class CenturyExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.getCenturyOfEra());
}
}
- private class DayExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class DayExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.dayOfMonth);
}
}
- private class DecadeExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class DecadeExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
- return DatumFactory.createFloat8((double) (tm.years / 10));
+ return DatumFactory.createFloat8((tm.years / 10.0d));
}
}
- private class DowExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class DowExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
Integer tdow = tm.getDayOfWeek();
@@ -167,84 +167,84 @@ public class DatePartFromTimestamp extends GeneralFunction {
}
}
- private class DoyExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class DoyExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double)tm.getDayOfYear());
}
}
- private class EpochExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class EpochExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double)DateTimeUtil.julianTimeToEpoch(DateTimeUtil.toJulianTimestamp(tm)));
}
}
- private class HourExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class HourExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.hours);
}
}
- private class ISODowExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class ISODowExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.getISODayOfWeek());
}
}
- private class ISOYearExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class ISOYearExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.getWeekyear());
}
}
- private class MicrosecondsExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class MicrosecondsExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) (tm.secs * 1000000 + tm.fsecs));
}
}
- private class MillenniumExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class MillenniumExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) (((tm.years - 1) / 1000) + 1));
}
}
- private class MillisecondsExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class MillisecondsExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) (tm.secs * 1000 + tm.fsecs / 1000.0));
}
}
- private class MinuteExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class MinuteExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.minutes);
}
}
- private class MonthExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class MonthExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.monthOfYear);
}
}
- private class QuarterExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class QuarterExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) (((tm.monthOfYear - 1) / 3) + 1));
}
}
- private class SecondExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class SecondExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
if (tm.fsecs != 0) {
@@ -255,21 +255,21 @@ public class DatePartFromTimestamp extends GeneralFunction {
}
}
- private class WeekExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class WeekExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.getWeekOfYear());
}
}
- private class YearExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class YearExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return DatumFactory.createFloat8((double) tm.years);
}
}
- private class NullExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
+ private static class NullExtractorFromTimestamp implements DatePartExtractorFromTimestamp {
@Override
public Datum extract(TimeMeta tm) {
return NullDatum.get();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
index 4e0449d..6bf99dd 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
@@ -104,35 +104,35 @@ public class DateTimePartFromUnixTimestamp extends GeneralFunction {
public Datum extract(TimeMeta dateTime, int week);
}
- private class DayExtractorFromTime implements DateTimePartExtractorFromUnixTime {
+ private static class DayExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getDay(dateTime));
}
}
- private class HourExtractorFromTime implements DateTimePartExtractorFromUnixTime {
+ private static class HourExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getHour(dateTime));
}
}
- private class MonthExtractorFromTime implements DateTimePartExtractorFromUnixTime {
+ private static class MonthExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getMonth(dateTime));
}
}
- private class YearExtractorFromTime implements DateTimePartExtractorFromUnixTime {
+ private static class YearExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getYear(dateTime));
}
}
- private class WeekExtractorFromTime implements WeekPartExtractorFromUnixTime {
+ private static class WeekExtractorFromTime implements WeekPartExtractorFromUnixTime {
@Override
public Datum extract(TimeMeta dateTime , int week) {
return DatumFactory.createInt8(DateTimeUtil.getDayOfWeek(dateTime,week));
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java
index 6b888ec..e3abeb5 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/string/RegexpReplace.java
@@ -57,9 +57,7 @@ public class RegexpReplace extends GeneralFunction {
@Expose protected boolean isPatternConstant;
// transient variables
- protected String pattern;
private boolean isAlwaysNull = false;
- private BooleanDatum result;
protected Pattern compiled;
public RegexpReplace() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java
index 93cdffb..ba3b3b6 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/FirstValue.java
@@ -56,7 +56,7 @@ public abstract class FirstValue extends WindowAggFunc<Datum> {
}
}
- protected class FirstValueContext implements FunctionContext {
+ protected static class FirstValueContext implements FunctionContext {
boolean isSet = false;
Datum first = null;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java
index 286cf03..5107297 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Lag.java
@@ -75,7 +75,7 @@ public abstract class Lag extends WindowAggFunc<Datum> {
}
}
- protected class LagContext implements FunctionContext {
+ protected static class LagContext implements FunctionContext {
CircularFifoBuffer lagBuffer = null;
Datum defaultDatum = null;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java
index 0df3f23..e469c83 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java
@@ -77,7 +77,7 @@ public final class Rank extends WindowAggFunc {
return new RankContext();
}
- private class RankContext implements FunctionContext {
+ private static class RankContext implements FunctionContext {
long rank = 0;
long accumulatedCount = 1;
Datum [] latest = null;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java
index bd838f8..212b597 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java
@@ -61,7 +61,7 @@ public class RowNumber extends WindowAggFunc<Datum> {
return DatumFactory.createInt8(((RowNumberContext) ctx).count);
}
- protected class RowNumberContext implements FunctionContext {
+ protected static class RowNumberContext implements FunctionContext {
long count = 0;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/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 d043a27..ac1c9ad 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
@@ -912,21 +912,19 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
&& ((PartitionedTableScanNode)scanNode).getInputPaths() != null &&
((PartitionedTableScanNode)scanNode).getInputPaths().length > 0) {
- if (scanNode instanceof PartitionedTableScanNode) {
- if (broadcastFlag) {
- PartitionedTableScanNode partitionedTableScanNode = (PartitionedTableScanNode) scanNode;
- List<Fragment> fileFragments = TUtil.newList();
- FileStorageManager fileStorageManager = (FileStorageManager)StorageManager.getFileStorageManager(ctx.getConf());
- for (Path path : partitionedTableScanNode.getInputPaths()) {
- fileFragments.addAll(TUtil.newList(fileStorageManager.split(scanNode.getCanonicalName(), path)));
- }
-
- FragmentProto[] fragments =
+ if (broadcastFlag) {
+ PartitionedTableScanNode partitionedTableScanNode = (PartitionedTableScanNode) scanNode;
+ List<Fragment> fileFragments = TUtil.newList();
+ FileStorageManager fileStorageManager = (FileStorageManager)StorageManager.getFileStorageManager(ctx.getConf());
+ for (Path path : partitionedTableScanNode.getInputPaths()) {
+ fileFragments.addAll(TUtil.newList(fileStorageManager.split(scanNode.getCanonicalName(), path)));
+ }
+
+ FragmentProto[] fragments =
FragmentConvertor.toFragmentProtoArray(fileFragments.toArray(new FileFragment[fileFragments.size()]));
- ctx.addFragments(scanNode.getCanonicalName(), fragments);
- return new PartitionMergeScanExec(ctx, scanNode, fragments);
- }
+ ctx.addFragments(scanNode.getCanonicalName(), fragments);
+ return new PartitionMergeScanExec(ctx, scanNode, fragments);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/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 0372769..9f82672 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
@@ -169,6 +169,14 @@ public class ExecutionBlockCursor {
}
return eb.equals(((BuildOrderItem) obj).eb);
}
+
+ @Override
+ public int hashCode() {
+ int result = eb != null ? eb.hashCode() : 0;
+ result = 31 * result + (parentEB != null ? parentEB.hashCode() : 0);
+ result = 31 * result + (siblings != null ? siblings.hashCode() : 0);
+ return result;
+ }
}
public boolean hasNext() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/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 6c3e3b8..f3dcf86 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
@@ -98,7 +98,7 @@ public class GlobalPlanner {
return storeType;
}
- public class GlobalPlanContext {
+ public static class GlobalPlanContext {
MasterPlan plan;
Map<Integer, ExecutionBlock> execBlockMap = Maps.newHashMap();
@@ -164,7 +164,7 @@ public class GlobalPlanner {
masterPlan.setTerminal(terminalBlock);
LOG.info("\n" + masterPlan.toString());
- masterPlan = rewriteEngine.rewrite(masterPlan);
+ rewriteEngine.rewrite(masterPlan);
}
private static void setFinalOutputChannel(DataChannel outputChannel, Schema outputSchema) {
@@ -1573,7 +1573,7 @@ public class GlobalPlanner {
}
@SuppressWarnings("unused")
- private class ConsecutiveUnionFinder extends BasicLogicalPlanVisitor<List<UnionNode>, LogicalNode> {
+ private static class ConsecutiveUnionFinder extends BasicLogicalPlanVisitor<List<UnionNode>, LogicalNode> {
@Override
public LogicalNode visitUnion(List<UnionNode> unionNodeList, LogicalPlan plan, LogicalPlan.QueryBlock queryBlock,
UnionNode node, Stack<LogicalNode> stack)
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/builder/DistinctGroupbyBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/builder/DistinctGroupbyBuilder.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/builder/DistinctGroupbyBuilder.java
index 5c6e80e..b5e9104 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/builder/DistinctGroupbyBuilder.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/builder/DistinctGroupbyBuilder.java
@@ -185,8 +185,6 @@ public class DistinctGroupbyBuilder {
private DistinctGroupbyNode buildMultiLevelBaseDistinctGroupByNode(GlobalPlanContext context,
ExecutionBlock latestExecBlock,
GroupbyNode groupbyNode) {
- LogicalPlan plan = context.getPlan().getLogicalPlan();
-
/*
Making DistinctGroupbyNode from GroupByNode
select col1, count(distinct col2), count(distinct col3), sum(col4) from ... group by col1
@@ -249,8 +247,7 @@ public class DistinctGroupbyBuilder {
}
//Add child groupby node for each Distinct clause
- for (String eachKey: distinctNodeBuildInfos.keySet()) {
- DistinctGroupbyNodeBuildInfo buildInfo = distinctNodeBuildInfos.get(eachKey);
+ for (DistinctGroupbyNodeBuildInfo buildInfo: distinctNodeBuildInfos.values()) {
GroupbyNode eachGroupbyNode = buildInfo.getGroupbyNode();
List<AggregationFunctionCallEval> groupbyAggFunctions = buildInfo.getAggFunctions();
String [] firstPhaseEvalNames = new String[groupbyAggFunctions.size()];
@@ -421,8 +418,7 @@ public class DistinctGroupbyBuilder {
}
//Add child groupby node for each Distinct clause
- for (String eachKey: distinctNodeBuildInfos.keySet()) {
- DistinctGroupbyNodeBuildInfo buildInfo = distinctNodeBuildInfos.get(eachKey);
+ for (DistinctGroupbyNodeBuildInfo buildInfo: distinctNodeBuildInfos.values()) {
GroupbyNode eachGroupbyNode = buildInfo.getGroupbyNode();
List<AggregationFunctionCallEval> groupbyAggFunctions = buildInfo.getAggFunctions();
Target[] targets = new Target[eachGroupbyNode.getGroupingColumns().length + groupbyAggFunctions.size()];
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
index 8c484a0..a4b9fe4 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
@@ -27,21 +27,15 @@ import org.apache.tajo.worker.TaskAttemptContext;
import java.io.IOException;
public abstract class AggregationExec extends UnaryPhysicalExec {
- protected GroupbyNode plan;
protected final int groupingKeyNum;
protected int groupingKeyIds[];
protected final int aggFunctionsNum;
protected final AggregationFunctionCallEval aggFunctions[];
- protected Schema evalSchema;
-
public AggregationExec(final TaskAttemptContext context, GroupbyNode plan,
PhysicalExec child) throws IOException {
super(context, plan.getInSchema(), plan.getOutSchema(), child);
- this.plan = plan;
-
- evalSchema = plan.getOutSchema();
final Column [] keyColumns = plan.getGroupingColumns();
groupingKeyNum = keyColumns.length;
@@ -68,6 +62,5 @@ public abstract class AggregationExec extends UnaryPhysicalExec {
@Override
public void close() throws IOException {
super.close();
- plan = null;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BNLJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BNLJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BNLJoinExec.java
index 14cf567..8b76097 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BNLJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BNLJoinExec.java
@@ -52,7 +52,7 @@ public class BNLJoinExec extends BinaryPhysicalExec {
private Tuple outputTuple = null;
private Tuple rightNext = null;
- private final int TUPLE_SLOT_SIZE = 10000;
+ private final static int TUPLE_SLOT_SIZE = 10000;
// projection
private Projector projector;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/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 8ee4e2f..4481569 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
@@ -67,8 +67,14 @@ public abstract class ColPartitionStoreExec extends UnaryPhysicalExec {
this.plan = plan;
if (plan.getType() == NodeType.CREATE_TABLE) {
+ if (!(plan instanceof CreateTableNode)) {
+ throw new IllegalArgumentException("plan should be a CreateTableNode type.");
+ }
this.outSchema = ((CreateTableNode)plan).getTableSchema();
} else if (plan.getType() == NodeType.INSERT) {
+ if (!(plan instanceof InsertNode)) {
+ throw new IllegalArgumentException("plan should be a InsertNode type.");
+ }
this.outSchema = ((InsertNode)plan).getTableSchema();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
index aca4879..2c6cc7e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
@@ -233,7 +233,6 @@ public class DistinctGroupbyFirstAggregationExec extends PhysicalExec {
}
class NonDistinctHashAggregator {
- private GroupbyNode groupbyNode;
private int aggFunctionsNum;
private final AggregationFunctionCallEval aggFunctions[];
@@ -243,7 +242,6 @@ public class DistinctGroupbyFirstAggregationExec extends PhysicalExec {
private Tuple dummyTuple;
private NonDistinctHashAggregator(GroupbyNode groupbyNode) throws IOException {
- this.groupbyNode = groupbyNode;
nonDistinctAggrDatas = new HashMap<Tuple, FunctionContext[]>();
@@ -305,7 +303,6 @@ public class DistinctGroupbyFirstAggregationExec extends PhysicalExec {
}
class DistinctHashAggregator {
- private GroupbyNode groupbyNode;
// GroupingKey -> DistinctKey
private Map<Tuple, Set<Tuple>> distinctAggrDatas;
@@ -321,7 +318,6 @@ public class DistinctGroupbyFirstAggregationExec extends PhysicalExec {
private boolean aggregatorFinished = false;
public DistinctHashAggregator(GroupbyNode groupbyNode) throws IOException {
- this.groupbyNode = groupbyNode;
Set<Integer> groupingKeyIndexSet = new HashSet<Integer>();
for (Integer eachIndex: groupingKeyIndexes) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/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 37d61a9..d3178db 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
@@ -36,7 +36,6 @@ import java.util.*;
import java.util.Map.Entry;
public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
- private DistinctGroupbyNode plan;
private boolean finished = false;
private HashAggregator[] hashAggregators;
@@ -286,7 +285,6 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
@Override
public void close() throws IOException {
- plan = null;
if (hashAggregators != null) {
for (int i = 0; i < hashAggregators.length; i++) {
hashAggregators[i].close();
@@ -337,17 +335,11 @@ public class DistinctGroupbyHashAggregationExec extends PhysicalExec {
private final int aggFunctionsNum;
private final AggregationFunctionCallEval aggFunctions[];
- private Schema evalSchema;
-
- private GroupbyNode groupbyNode;
-
int tupleSize;
public HashAggregator(GroupbyNode groupbyNode) throws IOException {
- this.groupbyNode = groupbyNode;
hashTable = new HashMap<Tuple, Map<Tuple, FunctionContext[]>>(10000);
- evalSchema = groupbyNode.getOutSchema();
List<Integer> distinctGroupingKeyIdSet = new ArrayList<Integer>();
for (int i = 0; i < distinctGroupingKeyIds.length; i++) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java
index 3a84f98..9ff479b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySortAggregationExec.java
@@ -32,7 +32,6 @@ import org.apache.tajo.worker.TaskAttemptContext;
import java.io.IOException;
public class DistinctGroupbySortAggregationExec extends PhysicalExec {
- private DistinctGroupbyNode plan;
private SortAggregateExec[] aggregateExecs;
private boolean finished = false;
@@ -46,7 +45,6 @@ public class DistinctGroupbySortAggregationExec extends PhysicalExec {
public DistinctGroupbySortAggregationExec(final TaskAttemptContext context, DistinctGroupbyNode plan,
SortAggregateExec[] aggregateExecs) throws IOException {
super(context, plan.getInSchema(), plan.getOutSchema());
- this.plan = plan;
this.aggregateExecs = aggregateExecs;
this.groupbyNodeNum = plan.getSubPlans().size();
@@ -166,7 +164,6 @@ public class DistinctGroupbySortAggregationExec extends PhysicalExec {
@Override
public void close() throws IOException {
- plan = null;
if (aggregateExecs != null) {
for (SortAggregateExec eachExec: aggregateExecs) {
eachExec.close();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
index a76b91d..26f09da 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
@@ -117,9 +117,9 @@ public class DistinctGroupbyThirdAggregationExec extends UnaryPhysicalExec {
// but groupbyResultTupleIndex's column type is protobuf
int matchedIndex = -1;
- for (Column eachIndexColumn: groupbyResultTupleIndex.keySet()) {
- if (eachIndexColumn.getQualifiedName().equals(eachOutputColumn.getQualifiedName())) {
- matchedIndex = groupbyResultTupleIndex.get(eachIndexColumn);
+ for (Map.Entry<Column, Integer> entry: groupbyResultTupleIndex.entrySet()) {
+ if (entry.getKey().getQualifiedName().equals(eachOutputColumn.getQualifiedName())) {
+ matchedIndex = entry.getValue();
break;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
index c3f9d3d..b3ebfb2 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
@@ -633,7 +633,7 @@ public class ExternalSortExec extends SortExec {
/**
* Two-way merger scanner that reads two input sources and outputs one output tuples sorted in some order.
*/
- private class PairWiseMerger implements Scanner {
+ private static class PairWiseMerger implements Scanner {
private Scanner leftScan;
private Scanner rightScan;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
index 38728b5..d475b78 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
@@ -101,11 +101,9 @@ public class HashJoinExec extends BinaryPhysicalExec {
}
}
- long scanStartTime = 0;
public Tuple next() throws IOException {
if (first) {
loadRightToHashTable();
- scanStartTime = System.currentTimeMillis();
}
Tuple rightTuple;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/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 28974f9..f1e2fe5 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
@@ -138,7 +138,7 @@ public final class HashShuffleFileWriteExec extends UnaryPhysicalExec {
entry.getValue().clear();
}
- TableStats aggregated = (TableStats)child.getInputStats().clone();
+ TableStats aggregated = (TableStats) child.getInputStats().clone();
aggregated.setNumBytes(writtenBytes);
aggregated.setNumRows(numRows);
context.setResultStats(aggregated);
@@ -146,6 +146,9 @@ public final class HashShuffleFileWriteExec extends UnaryPhysicalExec {
partitionTuples.clear();
return null;
+ } catch (RuntimeException e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
throw new IOException(e);
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
index 5e80b8f..df1c09d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
@@ -60,7 +60,6 @@ public class RightOuterMergeJoinExec extends BinaryPhysicalExec {
// projection
private Projector projector;
- private int rightNumCols;
private int leftNumCols;
private int posRightTupleSlots = -1;
private int posLeftTupleSlots = -1;
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
index 15f17fd..c62027d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
@@ -81,10 +81,12 @@ public class SeqScanExec extends PhysicalExec {
if (plan.isBroadcastTable()) {
String pathNameKey = "";
if (fragments != null) {
+ StringBuilder stringBuilder = new StringBuilder();
for (FragmentProto f : fragments) {
Fragment fragement = FragmentConvertor.convert(context.getConf(), f);
- pathNameKey += fragement.getKey();
+ stringBuilder.append(fragement.getKey());
}
+ pathNameKey = stringBuilder.toString();
}
cacheKey = new TupleCacheKey(
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
index 1a7dff9..e2dbf7a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
@@ -67,8 +67,6 @@ public class WindowAggExec extends UnaryPhysicalExec {
private boolean [] endUnboundedFollowingFlags;
private boolean [] endCurrentRowFlags;
- private boolean endCurrentRow = false;
-
// operator state
enum WindowState {
NEW_WINDOW,
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheKey.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheKey.java b/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheKey.java
index 6f39d32..1cb01c2 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheKey.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheKey.java
@@ -44,6 +44,9 @@ public class TupleCacheKey {
@Override
public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
return toString().equals(o.toString());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java b/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java
index 743d70c..ba25172 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java
@@ -99,7 +99,7 @@ public class TupleCacheScanner implements Scanner {
if (cacheData.size() == 0) {
return 1.0f;
}
- return count / cacheData.size();
+ return ((float)count) / cacheData.size();
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/ha/HdfsServiceTracker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/ha/HdfsServiceTracker.java b/tajo-core/src/main/java/org/apache/tajo/ha/HdfsServiceTracker.java
index 1475a5d..4a782ec 100644
--- a/tajo-core/src/main/java/org/apache/tajo/ha/HdfsServiceTracker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/ha/HdfsServiceTracker.java
@@ -212,11 +212,16 @@ public class HdfsServiceTracker extends HAServiceTracker {
break;
case HAConstants.MASTER_INFO_ADDRESS:
address = conf.getSocketAddrVar(ConfVars.TAJO_MASTER_INFO_ADDRESS);
+ break;
default:
break;
}
- return NetUtils.createSocketAddr(masterName.split(":")[0] + ":" + address.getPort());
+ if (address != null) {
+ return NetUtils.createSocketAddr(masterName.split(":")[0] + ":" + address.getPort());
+ } else {
+ return null;
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/master/QueryInfo.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/QueryInfo.java b/tajo-core/src/main/java/org/apache/tajo/master/QueryInfo.java
index b11fd99..38e9403 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/QueryInfo.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/QueryInfo.java
@@ -237,4 +237,51 @@ public class QueryInfo implements GsonObject, History, Comparable<QueryInfo> {
public int compareTo(QueryInfo o) {
return queryId.compareTo(o.queryId);
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ QueryInfo queryInfo = (QueryInfo) o;
+
+ if (finishTime != queryInfo.finishTime) return false;
+ if (Float.compare(queryInfo.progress, progress) != 0) return false;
+ if (queryMasterClientPort != queryInfo.queryMasterClientPort) return false;
+ if (queryMasterInfoPort != queryInfo.queryMasterInfoPort) return false;
+ if (queryMasterPort != queryInfo.queryMasterPort) return false;
+ if (startTime != queryInfo.startTime) return false;
+ if (context != null ? !context.equals(queryInfo.context) : queryInfo.context != null) return false;
+ if (hostNameOfQM != null ? !hostNameOfQM.equals(queryInfo.hostNameOfQM) : queryInfo.hostNameOfQM != null)
+ return false;
+ if (jsonExpr != null ? !jsonExpr.equals(queryInfo.jsonExpr) : queryInfo.jsonExpr != null) return false;
+ if (lastMessage != null ? !lastMessage.equals(queryInfo.lastMessage) : queryInfo.lastMessage != null) return false;
+ if (queryId != null ? !queryId.equals(queryInfo.queryId) : queryInfo.queryId != null) return false;
+ if (queryIdStr != null ? !queryIdStr.equals(queryInfo.queryIdStr) : queryInfo.queryIdStr != null) return false;
+ if (queryState != queryInfo.queryState) return false;
+ if (resultDesc != null ? !resultDesc.equals(queryInfo.resultDesc) : queryInfo.resultDesc != null) return false;
+ if (sql != null ? !sql.equals(queryInfo.sql) : queryInfo.sql != null) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = queryId != null ? queryId.hashCode() : 0;
+ result = 31 * result + (context != null ? context.hashCode() : 0);
+ result = 31 * result + (sql != null ? sql.hashCode() : 0);
+ result = 31 * result + (queryState != null ? queryState.hashCode() : 0);
+ result = 31 * result + (progress != +0.0f ? Float.floatToIntBits(progress) : 0);
+ result = 31 * result + (int) (startTime ^ (startTime >>> 32));
+ result = 31 * result + (int) (finishTime ^ (finishTime >>> 32));
+ result = 31 * result + (lastMessage != null ? lastMessage.hashCode() : 0);
+ result = 31 * result + (hostNameOfQM != null ? hostNameOfQM.hashCode() : 0);
+ result = 31 * result + queryMasterPort;
+ result = 31 * result + queryMasterClientPort;
+ result = 31 * result + queryMasterInfoPort;
+ result = 31 * result + (queryIdStr != null ? queryIdStr.hashCode() : 0);
+ result = 31 * result + (resultDesc != null ? resultDesc.hashCode() : 0);
+ result = 31 * result + (jsonExpr != null ? jsonExpr.hashCode() : 0);
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoContainerProxy.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
index 996d356..7ed9fc5 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
@@ -60,7 +60,7 @@ public class TajoContainerProxy extends ContainerProxy {
}
@Override
- public void launch(ContainerLaunchContext containerLaunchContext) {
+ public synchronized void launch(ContainerLaunchContext containerLaunchContext) {
context.getResourceAllocator().addContainer(containerID, this);
this.hostName = container.getNodeId().getHost();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
index aee8a74..586abb0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -152,6 +152,9 @@ public class TajoMaster extends CompositeService {
@Override
public void serviceInit(Configuration _conf) throws Exception {
+ if (!(_conf instanceof TajoConf)) {
+ throw new IllegalArgumentException("_conf should be a TajoConf type.");
+ }
this.systemConf = (TajoConf) _conf;
Runtime.getRuntime().addShutdownHook(new Thread(new ShutdownHook()));
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/master/exec/NonForwardQueryResultSystemScanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/exec/NonForwardQueryResultSystemScanner.java b/tajo-core/src/main/java/org/apache/tajo/master/exec/NonForwardQueryResultSystemScanner.java
index f645dc5..93909d1 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/exec/NonForwardQueryResultSystemScanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/exec/NonForwardQueryResultSystemScanner.java
@@ -56,6 +56,7 @@ import org.apache.tajo.engine.planner.global.MasterPlan;
import org.apache.tajo.engine.planner.physical.PhysicalExec;
import org.apache.tajo.engine.query.QueryContext;
import org.apache.tajo.master.TajoMaster.MasterContext;
+import org.apache.tajo.plan.InvalidQueryException;
import org.apache.tajo.plan.LogicalPlan;
import org.apache.tajo.plan.PlanningException;
import org.apache.tajo.plan.expr.EvalNode;
@@ -120,7 +121,11 @@ public class NonForwardQueryResultSystemScanner implements NonForwardQueryResult
break;
}
}
-
+
+ if (leafBlock == null) {
+ throw new InvalidQueryException("Global planner could not find any leaf block.");
+ }
+
taskContext = new TaskAttemptContext(queryContext, null,
new TaskAttemptId(new TaskId(leafBlock.getId(), 0), 0),
null, null);
@@ -544,7 +549,7 @@ public class NonForwardQueryResultSystemScanner implements NonForwardQueryResult
@Override
public Tuple next() throws IOException {
- Tuple aTuple = null;
+ Tuple aTuple;
Tuple outTuple = new VTuple(outColumnNum);
if (isClosed) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java b/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
index 519aa9d..920fd39 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
@@ -216,8 +216,10 @@ public class TajoResourceTracker extends AbstractService implements TajoResource
synchronized(rmContext) {
for(int eachWorker: rmContext.getWorkers().keySet()) {
Worker worker = rmContext.getWorkers().get(eachWorker);
- WorkerResource resource = worker.getResource();
+
if(worker != null) {
+ WorkerResource resource = worker.getResource();
+
totalMemoryMB += resource.getMemoryMB();
totalAvailableMemoryMB += resource.getAvailableMemoryMB();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java b/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
index 3d28d85..8c5b96c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
@@ -100,4 +100,26 @@ public class TajoWorkerContainer extends TajoContainer {
public int compareTo(TajoContainer container) {
return getId().compareTo(container.getId());
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TajoWorkerContainer that = (TajoWorkerContainer) o;
+
+ if (id != null ? !id.equals(that.id) : that.id != null) return false;
+ if (nodeId != null ? !nodeId.equals(that.nodeId) : that.nodeId != null) return false;
+ if (worker != null ? !worker.equals(that.worker) : that.worker != null) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = id != null ? id.hashCode() : 0;
+ result = 31 * result + (nodeId != null ? nodeId.hashCode() : 0);
+ result = 31 * result + (worker != null ? worker.hashCode() : 0);
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/master/rm/Worker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/rm/Worker.java b/tajo-core/src/main/java/org/apache/tajo/master/rm/Worker.java
index edded4d..d426e80 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/rm/Worker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/rm/Worker.java
@@ -161,6 +161,37 @@ public class Worker implements EventHandler<WorkerEvent>, Comparable<Worker> {
return connectionInfo.compareTo(o.connectionInfo);
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ Worker worker = (Worker) o;
+
+ if (lastHeartbeatTime != worker.lastHeartbeatTime) return false;
+ if (connectionInfo != null ? !connectionInfo.equals(worker.connectionInfo) : worker.connectionInfo != null)
+ return false;
+ if (readLock != null ? !readLock.equals(worker.readLock) : worker.readLock != null) return false;
+ if (resource != null ? !resource.equals(worker.resource) : worker.resource != null) return false;
+ if (rmContext != null ? !rmContext.equals(worker.rmContext) : worker.rmContext != null) return false;
+ if (stateMachine != null ? !stateMachine.equals(worker.stateMachine) : worker.stateMachine != null) return false;
+ if (writeLock != null ? !writeLock.equals(worker.writeLock) : worker.writeLock != null) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = readLock != null ? readLock.hashCode() : 0;
+ result = 31 * result + (writeLock != null ? writeLock.hashCode() : 0);
+ result = 31 * result + (rmContext != null ? rmContext.hashCode() : 0);
+ result = 31 * result + (int) (lastHeartbeatTime ^ (lastHeartbeatTime >>> 32));
+ result = 31 * result + (resource != null ? resource.hashCode() : 0);
+ result = 31 * result + (connectionInfo != null ? connectionInfo.hashCode() : 0);
+ result = 31 * result + (stateMachine != null ? stateMachine.hashCode() : 0);
+ return result;
+ }
+
public static class AddNodeTransition implements SingleArcTransition<Worker, WorkerEvent> {
@Override
public void transition(Worker worker, WorkerEvent workerEvent) {
@@ -177,6 +208,9 @@ public class Worker implements EventHandler<WorkerEvent>, Comparable<Worker> {
@Override
public WorkerState transition(Worker worker, WorkerEvent event) {
+ if (!(event instanceof WorkerStatusEvent)) {
+ throw new IllegalArgumentException("event should be a WorkerStatusEvent type.");
+ }
WorkerStatusEvent statusEvent = (WorkerStatusEvent) event;
// TODO - the synchronization scope using rmContext is too coarsen.
@@ -212,6 +246,9 @@ public class Worker implements EventHandler<WorkerEvent>, Comparable<Worker> {
@Override
public void transition(Worker worker, WorkerEvent workerEvent) {
+ if (!(workerEvent instanceof WorkerReconnectEvent)) {
+ throw new IllegalArgumentException("workerEvent should be a WorkerReconnectEvent type.");
+ }
WorkerReconnectEvent castedEvent = (WorkerReconnectEvent) workerEvent;
Worker newWorker = castedEvent.getWorker();
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
index 060e620..c2740e5 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
@@ -403,6 +403,9 @@ public class Query implements EventHandler<QueryEvent> {
@Override
public QueryState transition(Query query, QueryEvent queryEvent) {
+ if (!(queryEvent instanceof QueryCompletedEvent)) {
+ throw new IllegalArgumentException("queryEvent should be a QueryCompletedEvent type.");
+ }
QueryCompletedEvent stageEvent = (QueryCompletedEvent) queryEvent;
QueryState finalState;
@@ -463,7 +466,7 @@ public class Query implements EventHandler<QueryEvent> {
ExecutionBlockId finalExecBlockId, Path finalOutputDir) throws Exception;
}
- private class QueryHookExecutor {
+ private static class QueryHookExecutor {
private List<QueryHook> hookList = TUtil.newList();
private QueryMaster.QueryMasterContext context;
@@ -485,7 +488,7 @@ public class Query implements EventHandler<QueryEvent> {
}
}
- private class MaterializedResultHook implements QueryHook {
+ private static class MaterializedResultHook implements QueryHook {
@Override
public boolean isEligible(QueryContext queryContext, Query query, ExecutionBlockId finalExecBlockId,
@@ -521,7 +524,7 @@ public class Query implements EventHandler<QueryEvent> {
}
}
- private class CreateTableHook implements QueryHook {
+ private static class CreateTableHook implements QueryHook {
@Override
public boolean isEligible(QueryContext queryContext, Query query, ExecutionBlockId finalExecBlockId,
@@ -560,7 +563,7 @@ public class Query implements EventHandler<QueryEvent> {
}
}
- private class InsertTableHook implements QueryHook {
+ private static class InsertTableHook implements QueryHook {
@Override
public boolean isEligible(QueryContext queryContext, Query query, ExecutionBlockId finalExecBlockId,
@@ -638,6 +641,9 @@ public class Query implements EventHandler<QueryEvent> {
@Override
public void transition(Query query, QueryEvent event) {
+ if (!(event instanceof StageCompletedEvent)) {
+ throw new IllegalArgumentException("event should be a StageCompletedEvent type.");
+ }
try {
query.completedStagesCount++;
StageCompletedEvent castEvent = (StageCompletedEvent) event;
@@ -674,6 +680,9 @@ public class Query implements EventHandler<QueryEvent> {
private static class DiagnosticsUpdateTransition implements SingleArcTransition<Query, QueryEvent> {
@Override
public void transition(Query query, QueryEvent event) {
+ if (!(event instanceof QueryDiagnosticsUpdateEvent)) {
+ throw new IllegalArgumentException("event should be a QueryDiagnosticsUpdateEvent type.");
+ }
query.addDiagnostic(((QueryDiagnosticsUpdateEvent) event).getDiagnosticUpdate());
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/64e47a40/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
index c3899d6..bfaaca8 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
@@ -99,6 +99,9 @@ public class QueryMaster extends CompositeService implements EventHandler {
public void init(Configuration conf) {
LOG.info("QueryMaster init");
+ if (!(conf instanceof TajoConf)) {
+ throw new IllegalArgumentException("conf should be a TajoConf type");
+ }
try {
this.systemConf = (TajoConf)conf;
this.connPool = RpcConnectionPool.getPool();
@@ -192,6 +195,9 @@ public class QueryMaster extends CompositeService implements EventHandler {
TajoWorkerProtocol.TajoWorkerProtocolService tajoWorkerProtocolService = rpc.getStub();
tajoWorkerProtocolService.cleanupExecutionBlocks(null, executionBlockListProto, NullCallback.get());
+ } catch (RuntimeException e) {
+ LOG.warn("Ignoring RuntimeException. " + e.getMessage(), e);
+ continue;
} catch (Exception e) {
continue;
} finally {
@@ -512,7 +518,7 @@ public class QueryMaster extends CompositeService implements EventHandler {
break;
}
try {
- long expireTime = System.currentTimeMillis() - expireIntervalTime * 60 * 1000;
+ long expireTime = System.currentTimeMillis() - expireIntervalTime * 60 * 1000l;
cleanExpiredFinishedQueryMasterTask(expireTime);
} catch (Exception e) {
LOG.error(e.getMessage(), e);