You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2013/09/09 04:23:36 UTC
[1/3] TAJO-149: Eliminate QueryConf and its file write. (hyunsik)
Updated Branches:
refs/heads/master e448b5efc -> 27f76811c
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
index 5fc6c46..79481fa 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
@@ -76,12 +76,13 @@ public class StorageManager {
public StorageManager(TajoConf conf) throws IOException {
this.conf = conf;
this.baseDir = new Path(conf.getVar(ConfVars.ROOT_DIR));
- this.tableBaseDir = new Path(this.baseDir, TajoConstants.WAREHOUSE_DIR);
+ this.tableBaseDir = new Path(this.baseDir, TajoConstants.WAREHOUSE_DIR_NAME);
this.fs = baseDir.getFileSystem(conf);
this.blocksMetadataEnabled = conf.getBoolean(DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED,
DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED_DEFAULT);
- if (!this.blocksMetadataEnabled)
+ if (!this.blocksMetadataEnabled) {
LOG.warn("does not support block metadata. ('dfs.datanode.hdfs-blocks-metadata.enabled')");
+ }
}
public static StorageManager get(TajoConf conf) throws IOException {
@@ -327,15 +328,6 @@ public class StorageManager {
return tablets;
}
- public void writeTableMeta(Path tableRoot, TableMeta meta)
- throws IOException {
- FileSystem fs = tableRoot.getFileSystem(conf);
- FSDataOutputStream out = fs.create(new Path(tableRoot, ".meta"));
- FileUtil.writeProto(out, meta.getProto());
- out.flush();
- out.close();
- }
-
public long calculateSize(Path tablePath) throws IOException {
FileSystem fs = tablePath.getFileSystem(conf);
long totalSize = 0;
[2/3] TAJO-149: Eliminate QueryConf and its file write. (hyunsik)
Posted by hy...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
index d1e92be..c4b6ae0 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
@@ -18,28 +18,29 @@
package org.apache.tajo.master.querymaster;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.Clock;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.state.*;
import org.apache.tajo.ExecutionBlockId;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.QueryId;
+import org.apache.tajo.TajoConstants;
import org.apache.tajo.TajoProtos.QueryState;
import org.apache.tajo.catalog.CatalogService;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.TableDescImpl;
-import org.apache.tajo.engine.planner.PlannerUtil;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.planner.global.MasterPlan;
-import org.apache.tajo.engine.planner.logical.NodeType;
-import org.apache.tajo.engine.planner.logical.StoreTableNode;
-import org.apache.tajo.master.ExecutionBlock;
import org.apache.tajo.master.ExecutionBlockCursor;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.master.event.*;
import org.apache.tajo.storage.StorageManager;
+import org.apache.tajo.util.TUtil;
import java.io.IOException;
import java.util.ArrayList;
@@ -54,14 +55,14 @@ public class Query implements EventHandler<QueryEvent> {
private static final Log LOG = LogFactory.getLog(Query.class);
// Facilities for Query
- private final QueryConf conf;
+ private final TajoConf systemConf;
private final Clock clock;
private String queryStr;
private Map<ExecutionBlockId, SubQuery> subqueries;
private final EventHandler eventHandler;
private final MasterPlan plan;
private final StorageManager sm;
- QueryMasterTask.QueryContext context;
+ QueryMasterTask.QueryMasterTaskContext context;
private ExecutionBlockCursor cursor;
// Query Status
@@ -108,13 +109,13 @@ public class Query implements EventHandler<QueryEvent> {
.installTopology();
- public Query(final QueryMasterTask.QueryContext context, final QueryId id,
+ public Query(final QueryMasterTask.QueryMasterTaskContext context, final QueryId id,
final long appSubmitTime,
final String queryStr,
final EventHandler eventHandler,
final MasterPlan plan) {
this.context = context;
- this.conf = context.getConf();
+ this.systemConf = context.getConf();
this.id = id;
this.clock = context.getClock();
this.appSubmitTime = appSubmitTime;
@@ -308,31 +309,21 @@ public class Query implements EventHandler<QueryEvent> {
} else { // Finish a query
if (query.checkQueryForCompleted() == QueryState.QUERY_SUCCEEDED) {
- SubQuery subQuery = query.getSubQuery(castEvent.getExecutionBlockId());
- TableDesc outputTableDesc = new TableDescImpl(query.context.getQueryMeta().getOutputTable(),
- subQuery.getTableMeta(), query.context.getQueryMeta().getOutputPath());
- query.setResultDesc(outputTableDesc);
-
- if (!query.context.getQueryMeta().isFileOutput()) {
- try {
- query.writeStat(query.context.getQueryMeta().getOutputPath(), subQuery);
- } catch (IOException e) {
- e.printStackTrace();
+
+ Path finalOutputDir = commitOutputData(query);
+ TableDesc finalTableDesc = buildOrUpdateResultTableDesc(query, castEvent.getExecutionBlockId(), finalOutputDir);
+
+ QueryContext queryContext = query.context.getQueryContext();
+ CatalogService catalog = query.context.getQueryMasterContext().getWorkerContext().getCatalog();
+
+ if (queryContext.hasOutputTable()) { // TRUE only if a query command is 'CREATE TABLE' OR 'INSERT INTO'
+ if (queryContext.isOutputOverwrite()) { // TRUE only if a query is 'INSERT OVERWRITE INTO'
+ catalog.deleteTable(finalOutputDir.getName());
}
+ catalog.addTable(finalTableDesc);
}
+ query.setResultDesc(finalTableDesc);
query.eventHandler.handle(new QueryFinishEvent(query.getId()));
-
- StoreTableNode storeTableNode = (StoreTableNode) PlannerUtil.findTopNode(subQuery.getBlock().getPlan(),
- NodeType.STORE);
- if (storeTableNode.isCreatedTable()) {
- query.context.getQueryMasterContext().getWorkerContext().getCatalog().addTable(outputTableDesc);
- } else if (storeTableNode.isOverwrite() && !query.context.getQueryMeta().isFileOutput()) {
- CatalogService catalog = query.context.getQueryMasterContext().getWorkerContext().getCatalog();
- TableDesc updatingTable = catalog.getTableDesc(outputTableDesc.getName());
- updatingTable.getMeta().setStat(outputTableDesc.getMeta().getStat());
- catalog.deleteTable(outputTableDesc.getName());
- catalog.addTable(updatingTable);
- }
}
return query.finished(QueryState.QUERY_SUCCEEDED);
@@ -342,14 +333,58 @@ public class Query implements EventHandler<QueryEvent> {
return QueryState.QUERY_FAILED;
}
}
- }
- private static class DiagnosticsUpdateTransition implements
- SingleArcTransition<Query, QueryEvent> {
- @Override
- public void transition(Query query, QueryEvent event) {
- query.addDiagnostic(((QueryDiagnosticsUpdateEvent) event)
- .getDiagnosticUpdate());
+ /**
+ * It moves a result data stored in a staging output dir into a final output dir.
+ */
+ public Path commitOutputData(Query query) {
+ QueryContext queryContext = query.context.getQueryContext();
+ Path stagingResultDir = new Path(queryContext.getStagingDir(), TajoConstants.RESULT_DIR_NAME);
+ Path finalOutputDir;
+ if (queryContext.hasOutputPath()) {
+ finalOutputDir = queryContext.getOutputPath();
+ try {
+ FileSystem fs = stagingResultDir.getFileSystem(query.systemConf);
+ fs.rename(stagingResultDir, finalOutputDir);
+ LOG.info("Moved from the staging dir to the output directory '" + finalOutputDir);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ } else {
+ finalOutputDir = new Path(queryContext.getStagingDir(), TajoConstants.RESULT_DIR_NAME);
+ }
+
+ return finalOutputDir;
+ }
+
+ /**
+ * It builds a table desc and update the table desc if necessary.
+ */
+ public TableDesc buildOrUpdateResultTableDesc(Query query, ExecutionBlockId finalExecBlockId, Path finalOutputDir) {
+ // Determine the output table name
+ SubQuery subQuery = query.getSubQuery(finalExecBlockId);
+ QueryContext queryContext = query.context.getQueryContext();
+ String outputTableName;
+ if (queryContext.hasOutputTable()) { // CREATE TABLE or INSERT STATEMENT
+ outputTableName = queryContext.getOutputTable();
+ } else { // SELECT STATEMENT
+ outputTableName = query.getId().toString();
+ }
+
+ TableDesc outputTableDesc = new TableDescImpl(outputTableName, subQuery.getTableMeta(), finalOutputDir);
+ TableDesc finalTableDesc = outputTableDesc;
+
+ // If a query has a target table, a TableDesc is updated.
+ if (queryContext.hasOutputTable()) { // CREATE TABLE or INSERT STATEMENT
+ if (queryContext.isOutputOverwrite()) {
+ CatalogService catalog = query.context.getQueryMasterContext().getWorkerContext().getCatalog();
+ Preconditions.checkNotNull(catalog, "CatalogService is NULL");
+ TableDesc updatingTable = catalog.getTableDesc(outputTableDesc.getName());
+ updatingTable.getMeta().setStat(outputTableDesc.getMeta().getStat());
+ finalTableDesc = updatingTable;
+ }
+ }
+ return finalTableDesc;
}
}
@@ -415,9 +450,32 @@ public class Query implements EventHandler<QueryEvent> {
}
}
- private void writeStat(Path outputPath, SubQuery subQuery)
- throws IOException {
- ExecutionBlock execBlock = subQuery.getBlock();
- sm.writeTableMeta(outputPath, subQuery.getTableMeta());
+ public static interface QueryHook {
+ QueryState getTargetState();
+ void onEvent(Query query);
+ }
+
+ public static class QueryHookManager {
+ Map<QueryState, List<QueryHook>> hookList = TUtil.newHashMap();
+
+ public void addHook(QueryHook hook) {
+ if (hookList.containsKey(hook.getTargetState())) {
+ hookList.get(hook.getTargetState()).add(hook);
+ } else {
+ hookList.put(hook.getTargetState(), TUtil.newList(hook));
+ }
+ }
+
+ public void doHooks(Query query) {
+ QueryState finalState = query.checkQueryForCompleted();
+ List<QueryHook> list = hookList.get(finalState);
+ if (list != null) {
+ for (QueryHook hook : list) {
+ hook.onEvent(query);
+ }
+ } else {
+ LOG.error("QueryHookManager cannot deal with " + finalState + " event");
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
index f6cf02e..809dce2 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
@@ -29,7 +29,7 @@ import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
import org.apache.tajo.ipc.TajoWorkerProtocol;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.master.TajoAsyncDispatcher;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.master.rm.WorkerResource;
@@ -46,7 +46,7 @@ public class QueryInProgress extends CompositeService {
private QueryId queryId;
- private QueryMeta queryMeta;
+ private QueryContext queryContext;
private TajoAsyncDispatcher dispatcher;
@@ -66,11 +66,11 @@ public class QueryInProgress extends CompositeService {
public QueryInProgress(
TajoMaster.MasterContext masterContext,
- QueryMeta queryMeta,
+ QueryContext queryContext,
QueryId queryId, String sql, LogicalRootNode plan) {
super(QueryInProgress.class.getName());
this.masterContext = masterContext;
- this.queryMeta = queryMeta;
+ this.queryContext = queryContext;
this.queryId = queryId;
this.plan = plan;
@@ -105,7 +105,7 @@ public class QueryInProgress extends CompositeService {
while(true) {
try {
if(masterContext.getResourceManager().isQueryMasterStopped(queryId)) {
- LOG.info("====> " + queryId + " QueryMaster stopped");
+ LOG.info(queryId + " QueryMaster stopped");
queryMasterStopped = true;
break;
}
@@ -206,7 +206,7 @@ public class QueryInProgress extends CompositeService {
null,
TajoWorkerProtocol.QueryExecutionRequestProto.newBuilder()
.setQueryId(queryId.getProto())
- .setQueryMeta(queryMeta.getProto())
+ .setQueryContext(queryContext.getProto())
.setLogicalPlanJson(PrimitiveProtos.StringProto.newBuilder().setValue(plan.toJson()).build())
.build(), NullCallback.get());
querySubmitted.set(true);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
index fb92616..424d5bf 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
@@ -28,7 +28,7 @@ import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
import org.apache.tajo.ipc.TajoMasterProtocol;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.master.rm.WorkerResource;
@@ -85,9 +85,9 @@ public class QueryJobManager extends CompositeService {
return dispatcher.getEventHandler();
}
- public QueryInfo createNewQueryJob(QueryMeta queryMeta, String sql, LogicalRootNode plan) throws Exception {
+ public QueryInfo createNewQueryJob(QueryContext queryContext, String sql, LogicalRootNode plan) throws Exception {
QueryId queryId = QueryIdFactory.newQueryId(masterContext.getResourceManager().getSeedQueryId());
- QueryInProgress queryInProgress = new QueryInProgress(masterContext,queryMeta, queryId, sql, plan);
+ QueryInProgress queryInProgress = new QueryInProgress(masterContext, queryContext, queryId, sql, plan);
synchronized(runningQueries) {
runningQueries.put(queryId, queryInProgress);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
index 53b9c05..d45988c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
@@ -21,16 +21,15 @@ package org.apache.tajo.master.querymaster;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.Clock;
import org.apache.hadoop.yarn.SystemClock;
import org.apache.hadoop.yarn.event.Event;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.hadoop.yarn.service.Service;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.planner.global.GlobalOptimizer;
import org.apache.tajo.ipc.TajoMasterProtocol;
import org.apache.tajo.master.GlobalPlanner;
@@ -62,7 +61,7 @@ public class QueryMaster extends CompositeService implements EventHandler {
private StorageManager storageManager;
- private QueryConf queryConf;
+ private TajoConf systemConf;
private Map<QueryId, QueryMasterTask> queryMasterTasks = new HashMap<QueryId, QueryMasterTask>();
@@ -84,20 +83,19 @@ public class QueryMaster extends CompositeService implements EventHandler {
public void init(Configuration conf) {
LOG.info("QueryMaster init");
try {
- queryConf = new QueryConf(conf);
- queryConf.addResource(new Path(QueryConf.QUERY_MASTER_FILENAME));
+ systemConf = (TajoConf)conf;
QUERY_SESSION_TIMEOUT = 60 * 1000;
- queryMasterContext = new QueryMasterContext(queryConf);
+ queryMasterContext = new QueryMasterContext(systemConf);
clock = new SystemClock();
this.dispatcher = new TajoAsyncDispatcher("querymaster_" + System.currentTimeMillis());
addIfService(dispatcher);
- this.storageManager = new StorageManager(queryConf);
+ this.storageManager = new StorageManager(systemConf);
- globalPlanner = new GlobalPlanner(queryConf, storageManager, dispatcher.getEventHandler());
+ globalPlanner = new GlobalPlanner(systemConf, storageManager, dispatcher.getEventHandler());
globalOptimizer = new GlobalOptimizer();
dispatcher.register(QueryStartEvent.EventType.class, new QueryStartEventHandler());
@@ -188,13 +186,13 @@ public class QueryMaster extends CompositeService implements EventHandler {
}
public class QueryMasterContext {
- private QueryConf conf;
+ private TajoConf conf;
- public QueryMasterContext(QueryConf conf) {
+ public QueryMasterContext(TajoConf conf) {
this.conf = conf;
}
- public QueryConf getConf() {
+ public TajoConf getConf() {
return conf;
}
@@ -255,9 +253,9 @@ public class QueryMaster extends CompositeService implements EventHandler {
LOG.info("Start QueryStartEventHandler:" + event.getQueryId());
//To change body of implemented methods use File | Settings | File Templates.
QueryMasterTask queryMasterTask = new QueryMasterTask(queryMasterContext,
- event.getQueryId(), event.getQueryMeta(), event.getLogicalPlanJson());
+ event.getQueryId(), event.getQueryContext(), event.getLogicalPlanJson());
- queryMasterTask.init(queryConf);
+ queryMasterTask.init(systemConf);
queryMasterTask.start();
synchronized(queryMasterTasks) {
queryMasterTasks.put(event.getQueryId(), queryMasterTask);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
index 5a79464..ae1c1e5 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
@@ -25,8 +25,9 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.hadoop.yarn.util.RackResolver;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.QueryId;
+import org.apache.tajo.TajoConstants;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.util.TajoIdUtils;
import java.io.PrintWriter;
@@ -34,9 +35,10 @@ import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
+@Deprecated
public class QueryMasterRunner extends AbstractService {
private static final Log LOG = LogFactory.getLog(QueryMasterRunner.class);
- private QueryConf queryConf;
+ private TajoConf systemConf;
private QueryMaster queryMaster;
private QueryId queryId;
private String queryMasterManagerAddress;
@@ -59,11 +61,9 @@ public class QueryMasterRunner extends AbstractService {
@Override
public void init(Configuration conf) {
- this.queryConf = (QueryConf)conf;
- RackResolver.init(queryConf);
-
+ this.systemConf = (TajoConf)conf;
+ RackResolver.init(systemConf);
Runtime.getRuntime().addShutdownHook(new Thread(new ShutdownHook()));
-
super.init(conf);
}
@@ -72,7 +72,7 @@ public class QueryMasterRunner extends AbstractService {
//create QueryMaster
QueryMaster query = new QueryMaster(null);
- query.init(queryConf);
+ query.init(systemConf);
query.start();
}
@@ -83,8 +83,8 @@ public class QueryMasterRunner extends AbstractService {
public static void main(String[] args) throws Exception {
LOG.info("QueryMasterRunner started");
- final QueryConf conf = new QueryConf();
- conf.addResource(new Path(QueryConf.QUERY_MASTER_FILENAME));
+ final TajoConf conf = new TajoConf();
+ conf.addResource(new Path(TajoConstants.SYSTEM_CONF_FILENAME));
UserGroupInformation.setConfiguration(conf);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
index f348e44..2212f82 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
@@ -39,13 +39,12 @@ import org.apache.tajo.engine.planner.logical.LogicalNode;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
import org.apache.tajo.engine.planner.logical.NodeType;
import org.apache.tajo.engine.planner.logical.ScanNode;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.master.TajoAsyncDispatcher;
import org.apache.tajo.master.event.*;
import org.apache.tajo.master.rm.TajoWorkerResourceManager;
import org.apache.tajo.rpc.NullCallback;
import org.apache.tajo.storage.StorageManager;
-import org.apache.tajo.storage.StorageUtil;
import org.apache.tajo.worker.AbstractResourceAllocator;
import org.apache.tajo.worker.TajoResourceAllocator;
import org.apache.tajo.worker.YarnResourceAllocator;
@@ -60,15 +59,15 @@ public class QueryMasterTask extends CompositeService {
private static final Log LOG = LogFactory.getLog(QueryMasterTask.class.getName());
// query submission directory is private!
- final public static FsPermission USER_DIR_PERMISSION =
+ final public static FsPermission STAGING_DIR_PERMISSION =
FsPermission.createImmutable((short) 0700); // rwx--------
private QueryId queryId;
- private QueryMeta queryMeta;
-
private QueryContext queryContext;
+ private QueryMasterTaskContext queryTaskContext;
+
private QueryMaster.QueryMasterContext queryMasterContext;
private Query query;
@@ -81,11 +80,9 @@ public class QueryMasterTask extends CompositeService {
private final long querySubmitTime;
- private Path outputPath;
-
private Map<String, TableDesc> tableDescMap = new HashMap<String, TableDesc>();
- private QueryConf queryConf;
+ private TajoConf systemConf;
private AtomicLong lastClientHeartbeat = new AtomicLong(-1);
@@ -94,28 +91,28 @@ public class QueryMasterTask extends CompositeService {
private AtomicBoolean stopped = new AtomicBoolean(false);
public QueryMasterTask(QueryMaster.QueryMasterContext queryMasterContext,
- QueryId queryId, QueryMeta queryMeta, String logicalPlanJson) {
+ QueryId queryId, QueryContext queryContext, String logicalPlanJson) {
super(QueryMasterTask.class.getName());
this.queryMasterContext = queryMasterContext;
this.queryId = queryId;
- this.queryMeta = queryMeta;
+ this.queryContext = queryContext;
this.logicalPlanJson = logicalPlanJson;
this.querySubmitTime = System.currentTimeMillis();
}
@Override
public void init(Configuration conf) {
- queryConf = new QueryConf(conf);
- queryConf.addResource(new Path(QueryConf.QUERY_MASTER_FILENAME));
+ systemConf = (TajoConf)conf;
+
try {
- queryContext = new QueryContext();
+ queryTaskContext = new QueryMasterTaskContext();
String resourceManagerClassName = conf.get("tajo.resource.manager",
TajoWorkerResourceManager.class.getCanonicalName());
if(resourceManagerClassName.indexOf(TajoWorkerResourceManager.class.getName()) >= 0) {
- resourceAllocator = new TajoResourceAllocator(queryContext);
+ resourceAllocator = new TajoResourceAllocator(queryTaskContext);
} else {
- resourceAllocator = new YarnResourceAllocator(queryContext);
+ resourceAllocator = new YarnResourceAllocator(queryTaskContext);
}
addService(resourceAllocator);
@@ -130,7 +127,7 @@ public class QueryMasterTask extends CompositeService {
initStagingDir();
- super.init(queryConf);
+ super.init(systemConf);
} catch (IOException e) {
LOG.error(e.getMessage(), e);
}
@@ -242,14 +239,14 @@ public class QueryMasterTask extends CompositeService {
MasterPlan globalPlan = queryMasterContext.getGlobalPlanner().build(queryId, logicalNodeRoot);
this.masterPlan = queryMasterContext.getGlobalOptimizer().optimize(globalPlan);
- query = new Query(queryContext, queryId, querySubmitTime,
- "", queryContext.getEventHandler(), masterPlan);
+ query = new Query(queryTaskContext, queryId, querySubmitTime,
+ "", queryTaskContext.getEventHandler(), masterPlan);
dispatcher.register(QueryEventType.class, query);
- queryContext.getEventHandler().handle(new QueryEvent(queryId,
+ queryTaskContext.getEventHandler().handle(new QueryEvent(queryId,
QueryEventType.INIT));
- queryContext.getEventHandler().handle(new QueryEvent(queryId,
+ queryTaskContext.getEventHandler().handle(new QueryEvent(queryId,
QueryEventType.START));
} catch (Exception e) {
LOG.error(e.getMessage(), e);
@@ -264,88 +261,85 @@ public class QueryMasterTask extends CompositeService {
* them to variables.
*/
private void initStagingDir() throws IOException {
+
String realUser;
String currentUser;
UserGroupInformation ugi;
ugi = UserGroupInformation.getLoginUser();
realUser = ugi.getShortUserName();
currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
+ FileSystem defaultFS = FileSystem.get(systemConf);
- String givenOutputTableName = queryMeta.getOutputTable();
- Path stagingDir;
+ Path stagingDir = null;
+ Path outputDir = null;
+ try {
+ ////////////////////////////////////////////
+ // Create Output Directory
+ ////////////////////////////////////////////
- // If final output directory is not given by an user,
- // we use the query id as a output directory.
- if (givenOutputTableName == null || givenOutputTableName.isEmpty()) {
- FileSystem defaultFS = FileSystem.get(queryConf);
+ stagingDir = new Path(TajoConf.getStagingRoot(systemConf), queryId.toString());
- Path homeDirectory = defaultFS.getHomeDirectory();
- if (!defaultFS.exists(homeDirectory)) {
- defaultFS.mkdirs(homeDirectory, new FsPermission(USER_DIR_PERMISSION));
+ if (defaultFS.exists(stagingDir)) {
+ throw new IOException("The staging directory '" + stagingDir + "' already exists");
}
-
- Path userQueryDir = new Path(homeDirectory, TajoConstants.USER_QUERYDIR_PREFIX);
-
- if (defaultFS.exists(userQueryDir)) {
- FileStatus fsStatus = defaultFS.getFileStatus(userQueryDir);
- String owner = fsStatus.getOwner();
-
- if (!(owner.equals(currentUser) || owner.equals(realUser))) {
- throw new IOException("The ownership on the user's query " +
- "directory " + userQueryDir + " is not as expected. " +
- "It is owned by " + owner + ". The directory must " +
- "be owned by the submitter " + currentUser + " or " +
- "by " + realUser);
- }
-
- if (!fsStatus.getPermission().equals(USER_DIR_PERMISSION)) {
- LOG.info("Permissions on staging directory " + userQueryDir + " are " +
- "incorrect: " + fsStatus.getPermission() + ". Fixing permissions " +
- "to correct value " + USER_DIR_PERMISSION);
- defaultFS.setPermission(userQueryDir, new FsPermission(USER_DIR_PERMISSION));
- }
- } else {
- defaultFS.mkdirs(userQueryDir,
- new FsPermission(USER_DIR_PERMISSION));
+ defaultFS.mkdirs(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
+ FileStatus fsStatus = defaultFS.getFileStatus(stagingDir);
+ String owner = fsStatus.getOwner();
+
+ if (!(owner.equals(currentUser) || owner.equals(realUser))) {
+ throw new IOException("The ownership on the user's query " +
+ "directory " + stagingDir + " is not as expected. " +
+ "It is owned by " + owner + ". The directory must " +
+ "be owned by the submitter " + currentUser + " or " +
+ "by " + realUser);
}
- stagingDir = StorageUtil.concatPath(userQueryDir, queryId.toString());
-
- if (defaultFS.exists(stagingDir)) {
- throw new IOException("The staging directory " + stagingDir
- + "already exists. The directory must be unique to each query");
- } else {
- defaultFS.mkdirs(stagingDir, new FsPermission(USER_DIR_PERMISSION));
+ if (!fsStatus.getPermission().equals(STAGING_DIR_PERMISSION)) {
+ LOG.info("Permissions on staging directory " + stagingDir + " are " +
+ "incorrect: " + fsStatus.getPermission() + ". Fixing permissions " +
+ "to correct value " + STAGING_DIR_PERMISSION);
+ defaultFS.setPermission(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
}
- // Set the query id to the output table name
- queryMeta.setOutputTable(queryId.toString());
-
- } else { // if a output table is given
-
- Path warehouseDir = new Path(queryConf.getVar(TajoConf.ConfVars.ROOT_DIR),
- TajoConstants.WAREHOUSE_DIR);
- FileSystem fs = warehouseDir.getFileSystem(queryConf);
-
- if (queryMeta.isFileOutput()) {
- stagingDir = queryMeta.getOutputPath();
- } else {
- stagingDir = new Path(warehouseDir, queryMeta.getOutputTable());
+ // Create a subdirectories
+ defaultFS.mkdirs(new Path(stagingDir, TajoConstants.RESULT_DIR_NAME));
+ LOG.info("The staging dir '" + outputDir + "' is created.");
+ queryContext.setStagingDir(stagingDir);
+
+ ////////////////////////////////////////////////////
+ // Check and Create An Output Directory If Necessary
+ ////////////////////////////////////////////////////
+ if (queryContext.hasOutputPath()) {
+ outputDir = queryContext.getOutputPath();
+ if (queryContext.isOutputOverwrite()) {
+ if (defaultFS.exists(outputDir.getParent())) {
+ if (defaultFS.exists(outputDir)) {
+ defaultFS.delete(outputDir, true);
+ LOG.info("The output directory '" + outputDir + "' is cleaned.");
+ }
+ } else {
+ defaultFS.mkdirs(outputDir.getParent());
+ LOG.info("The output directory's parent '" + outputDir.getParent() + "' is created.");
+ }
+ } else {
+ if (defaultFS.exists(outputDir)) {
+ throw new IOException("The output directory '" + outputDir + " already exists.");
+ }
+ }
+ }
+ } catch (IOException ioe) {
+ if (stagingDir != null && defaultFS.exists(stagingDir)) {
+ defaultFS.delete(stagingDir, true);
+ LOG.info("The staging directory '" + stagingDir + "' is deleted");
}
- if (!queryMeta.isOutputOverwrite()) {
- if (fs.exists(stagingDir)) {
- throw new IOException("The staging directory " + stagingDir
- + " already exists. The directory must be unique to each query");
- }
+ if (outputDir != null && defaultFS.exists(outputDir)) {
+ defaultFS.delete(outputDir, true);
+ LOG.info("The output directory '" + outputDir + "' is deleted");
}
- fs.mkdirs(stagingDir, new FsPermission(USER_DIR_PERMISSION));
+ throw ioe;
}
-
- queryMeta.setOutputPath(stagingDir);
- outputPath = stagingDir;
- LOG.info("Initialized Query Staging Dir: " + outputPath);
}
public Query getQuery() {
@@ -356,12 +350,12 @@ public class QueryMasterTask extends CompositeService {
stop();
}
- public QueryContext getQueryContext() {
- return queryContext;
+ public QueryMasterTaskContext getQueryTaskContext() {
+ return queryTaskContext;
}
public EventHandler getEventHandler() {
- return queryContext.getEventHandler();
+ return queryTaskContext.getEventHandler();
}
public void touchSessionTime() {
@@ -384,18 +378,18 @@ public class QueryMasterTask extends CompositeService {
}
}
- public class QueryContext {
+ public class QueryMasterTaskContext {
EventHandler eventHandler;
public QueryMaster.QueryMasterContext getQueryMasterContext() {
return queryMasterContext;
}
- public QueryMeta getQueryMeta() {
- return queryMeta;
+ public QueryContext getQueryContext() {
+ return queryContext;
}
- public QueryConf getConf() {
- return queryConf;
+ public TajoConf getConf() {
+ return systemConf;
}
public Clock getClock() {
@@ -414,8 +408,8 @@ public class QueryMasterTask extends CompositeService {
return queryMasterContext.getStorageManager();
}
- public Path getOutputPath() {
- return outputPath;
+ public Path getStagingDir() {
+ return queryContext.getStagingDir();
}
public synchronized EventHandler getEventHandler() {
@@ -445,7 +439,7 @@ public class QueryMasterTask extends CompositeService {
}
public AbstractResourceAllocator getResourceAllocator() {
- return (AbstractResourceAllocator)resourceAllocator;
+ return resourceAllocator;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
index 0cd5d64..43895d8 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
@@ -69,14 +69,14 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
private static final Log LOG = LogFactory.getLog(SubQuery.class);
- private QueryMeta queryMeta;
+ private QueryContext queryContext;
private ExecutionBlock block;
private int priority;
private TableMeta meta;
private EventHandler eventHandler;
private final StorageManager sm;
private TaskSchedulerImpl taskScheduler;
- private QueryMasterTask.QueryContext context;
+ private QueryMasterTask.QueryMasterTaskContext context;
private long startTime;
private long finishTime;
@@ -136,7 +136,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
private int completedTaskCount = 0;
- public SubQuery(QueryMasterTask.QueryContext context, ExecutionBlock block, StorageManager sm) {
+ public SubQuery(QueryMasterTask.QueryMasterTaskContext context, ExecutionBlock block, StorageManager sm) {
this.context = context;
this.block = block;
this.sm = sm;
@@ -153,7 +153,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
state == SubQueryState.CONTAINER_ALLOCATED || state == SubQueryState.RUNNING;
}
- public QueryMasterTask.QueryContext getContext() {
+ public QueryMasterTask.QueryMasterTaskContext getContext() {
return context;
}
@@ -335,9 +335,10 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
} else {
stat = computeStatFromTasks();
}
- TableMeta meta = writeStat(this, stat);
+
+ StoreTableNode storeTableNode = getBlock().getStoreTableNode();
+ TableMeta meta = toTableMeta(storeTableNode);
meta.setStat(stat);
- setTableMeta(meta);
return meta;
}
@@ -356,7 +357,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
StoreTableNode storeTableNode = execBlock.getStoreTableNode();
TableMeta meta = toTableMeta(storeTableNode);
meta.setStat(stat);
- sm.writeTableMeta(sm.getTablePath(execBlock.getOutputName()), meta);
+ //sm.writeTableMeta(sm.getTablePath(execBlock.getOutputName()), meta);
return meta;
}
@@ -597,7 +598,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
return maxTaskNum;
}
- public static long getInputVolume(QueryMasterTask.QueryContext context, ExecutionBlock execBlock) {
+ public static long getInputVolume(QueryMasterTask.QueryMasterTaskContext context, ExecutionBlock execBlock) {
Map<String, TableDesc> tableMap = context.getTableDescMap();
if (execBlock.isLeafBlock()) {
ScanNode outerScan = execBlock.getScanNodes()[0];
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
index 3701d58..eb710fa 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
@@ -211,15 +211,6 @@ public class TajoWorkerResourceManager implements WorkerResourceManager {
LOG.info("====> allocateWorkerResources: allocated:" + workerResources.size());
-// if(LOG.isDebugEnabled()) {
-// LOG.debug("====> allocateWorkerResources:" +
-// (new ExecutionBlockId(resourceRequest.request.getExecutionBlockId())) +
-// ", required:" + resourceRequest.request.getNumWorks() + ", allocated:" + workerResources.size());
-// } else {
-// LOG.info("====> allocateWorkerResources: required:" + resourceRequest.request.getNumWorks() +
-// ", allocated:" + workerResources.size() + ", queryMasterRequest=" + resourceRequest.queryMasterRequest);
-// }
-
if(workerResources.size() > 0) {
if(resourceRequest.queryMasterRequest) {
startQueryMaster(resourceRequest.queryId, workerResources.get(0));
@@ -386,7 +377,7 @@ public class TajoWorkerResourceManager implements WorkerResourceManager {
allWorkerResourceMap.put(workerResource.getId(), workerResource);
liveWorkerResources.add(hostAndPort);
- LOG.info("====> TajoWorker:" + workerResource + " added in live TajoWorker list");
+ LOG.info("TajoWorker:" + workerResource + " added in live TajoWorker list");
workerResourceLock.notifyAll();
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
index 4392158..2202183 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
@@ -57,10 +57,10 @@ public class YarnRMContainerAllocator extends AMRMClientImpl
private static final Log LOG = LogFactory.getLog(YarnRMContainerAllocator.
class.getName());
- private QueryMasterTask.QueryContext context;
+ private QueryMasterTask.QueryMasterTaskContext context;
private final EventHandler eventHandler;
- public YarnRMContainerAllocator(QueryMasterTask.QueryContext context) {
+ public YarnRMContainerAllocator(QueryMasterTask.QueryMasterTaskContext context) {
super(ApplicationIdUtils.createApplicationAttemptId(context.getQueryId()));
this.context = context;
this.eventHandler = context.getDispatcher().getEventHandler();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
index fe31bc6..da7daf0 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
@@ -246,7 +246,6 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
//vargs.add("-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5005");
//}
// Set class name
- //vargs.add(QueryMasterRunner.class.getCanonicalName());
vargs.add(TajoWorker.class.getCanonicalName());
vargs.add("qm");
vargs.add(queryId.toString()); // queryId
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
index c1afb80..fa2ff14 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
@@ -31,8 +31,6 @@ import org.apache.hadoop.yarn.api.records.impl.pb.NodeIdPBImpl;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.proto.YarnProtos;
import org.apache.tajo.ExecutionBlockId;
-import org.apache.tajo.QueryConf;
-import org.apache.tajo.TajoConstants;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.ipc.TajoMasterProtocol;
import org.apache.tajo.master.ContainerProxy;
@@ -68,15 +66,15 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
static AtomicInteger containerIdSeq = new AtomicInteger(0);
private TajoConf tajoConf;
- private QueryMasterTask.QueryContext queryContext;
+ private QueryMasterTask.QueryMasterTaskContext queryTaskContext;
private final ExecutorService executorService;
private AtomicBoolean stopped = new AtomicBoolean(false);
- public TajoResourceAllocator(QueryMasterTask.QueryContext queryContext) {
- this.queryContext = queryContext;
+ public TajoResourceAllocator(QueryMasterTask.QueryMasterTaskContext queryTaskContext) {
+ this.queryTaskContext = queryTaskContext;
executorService = Executors.newFixedThreadPool(
- queryContext.getConf().getIntVar(TajoConf.ConfVars.AM_TASKRUNNER_LAUNCH_PARALLEL_NUM));
+ queryTaskContext.getConf().getIntVar(TajoConf.ConfVars.AM_TASKRUNNER_LAUNCH_PARALLEL_NUM));
}
@Override
@@ -102,9 +100,9 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
public void init(Configuration conf) {
tajoConf = (TajoConf)conf;
- queryContext.getDispatcher().register(TaskRunnerGroupEvent.EventType.class, new TajoTaskRunnerLauncher());
+ queryTaskContext.getDispatcher().register(TaskRunnerGroupEvent.EventType.class, new TajoTaskRunnerLauncher());
//
- queryContext.getDispatcher().register(ContainerAllocatorEventType.class, new TajoWorkerAllocationHandler());
+ queryTaskContext.getDispatcher().register(ContainerAllocatorEventType.class, new TajoWorkerAllocationHandler());
super.init(conf);
}
@@ -117,7 +115,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
stopped.set(true);
executorService.shutdownNow();
- Map<ContainerId, ContainerProxy> containers = queryContext.getResourceAllocator().getContainers();
+ Map<ContainerId, ContainerProxy> containers = queryTaskContext.getResourceAllocator().getContainers();
List<ContainerProxy> list = new ArrayList<ContainerProxy>(containers.values());
for(ContainerProxy eachProxy: list) {
try {
@@ -162,36 +160,10 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
}
private void launchTaskRunners(ExecutionBlockId executionBlockId, Collection<Container> containers) {
- FileSystem fs = null;
-
- QueryConf queryConf = queryContext.getConf();
- LOG.info("defaultFS: " + queryConf.get("fs.default.name"));
- LOG.info("defaultFS: " + queryConf.get("fs.defaultFS"));
- try {
- fs = FileSystem.get(queryConf);
- } catch (IOException e) {
- LOG.error(e.getMessage(), e);
- }
-
- try {
- // TODO move to tajo temp
- Path warehousePath = new Path(queryConf.getVar(TajoConf.ConfVars.ROOT_DIR), TajoConstants.WAREHOUSE_DIR);
- Path queryConfPath = new Path(warehousePath, executionBlockId.getQueryId().toString());
- queryConfPath = new Path(queryConfPath, QueryConf.FILENAME);
-
- if(!fs.exists(queryConfPath)){
- LOG.info("Writing a QueryConf to HDFS and add to local environment, outputPath=" + queryConfPath);
- writeConf(queryConf, queryConfPath);
- } else {
- LOG.warn("QueryConf already exist. path: " + queryConfPath.toString());
- }
- } catch (IOException e) {
- LOG.error(e.getMessage(), e);
- }
- //Query in standby mode doesn't need launch Worker.
- //But, Assign ExecutionBlock to assigned tajo worker
+ // Query in standby mode doesn't need launch Worker.
+ // But, Assign ExecutionBlock to assigned tajo worker
for(Container eachContainer: containers) {
- TajoContainerProxy containerProxy = new TajoContainerProxy(queryContext, tajoConf,
+ TajoContainerProxy containerProxy = new TajoContainerProxy(queryTaskContext, tajoConf,
eachContainer, executionBlockId);
executorService.submit(new LaunchRunner(eachContainer.getId(), containerProxy));
}
@@ -213,7 +185,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
private void stopContainers(Collection<Container> containers) {
for (Container container : containers) {
- final ContainerProxy proxy = queryContext.getResourceAllocator().getContainer(container.getId());
+ final ContainerProxy proxy = queryTaskContext.getResourceAllocator().getContainer(container.getId());
executorService.submit(new StopContainerRunner(container.getId(), proxy));
}
}
@@ -248,7 +220,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
@Override
public void run() {
- LOG.info("======> Start TajoWorkerAllocationThread");
+ LOG.info("Start TajoWorkerAllocationThread");
CallFuture2<TajoMasterProtocol.WorkerResourceAllocationResponse> callBack =
new CallFuture2<TajoMasterProtocol.WorkerResourceAllocationResponse>();
@@ -262,7 +234,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
.setExecutionBlockId(event.getExecutionBlockId().getProto())
.build();
- queryContext.getQueryMasterContext().getWorkerContext().
+ queryTaskContext.getQueryMasterContext().getWorkerContext().
getTajoMasterRpcClient().allocateWorkerResources(null, request, callBack);
TajoMasterProtocol.WorkerResourceAllocationResponse response = null;
@@ -315,14 +287,14 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
containers.add(container);
}
- SubQueryState state = queryContext.getSubQuery(executionBlockId).getState();
+ SubQueryState state = queryTaskContext.getSubQuery(executionBlockId).getState();
if (!SubQuery.isRunningState(state)) {
List<WorkerResource> workerResources = new ArrayList<WorkerResource>();
for(Container eachContainer: containers) {
workerResources.add(((TajoWorkerContainer)eachContainer).getWorkerResource());
}
try {
- TajoContainerProxy.releaseWorkerResource(queryContext, executionBlockId, workerResources);
+ TajoContainerProxy.releaseWorkerResource(queryTaskContext, executionBlockId, workerResources);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
@@ -333,7 +305,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
if(LOG.isDebugEnabled()) {
LOG.debug("SubQueryContainerAllocationEvent fire:" + executionBlockId);
}
- queryContext.getEventHandler().handle(new SubQueryContainerAllocationEvent(executionBlockId, containers));
+ queryTaskContext.getEventHandler().handle(new SubQueryContainerAllocationEvent(executionBlockId, containers));
}
numAllocatedWorkers += workerHosts.size();
@@ -345,7 +317,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
event.getRequiredNum() - numAllocatedWorkers,
event.isLeafQuery(), event.getProgress()
);
- queryContext.getEventHandler().handle(shortRequestEvent);
+ queryTaskContext.getEventHandler().handle(shortRequestEvent);
}
LOG.info("Stop TajoWorkerAllocationThread");
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
index 0fc896b..ff25fb3 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -21,13 +21,16 @@ package org.apache.tajo.worker;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.hadoop.yarn.util.RackResolver;
import org.apache.tajo.QueryId;
+import org.apache.tajo.TajoConstants;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.catalog.CatalogClient;
+import org.apache.tajo.catalog.CatalogConstants;
import org.apache.tajo.catalog.CatalogService;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.ipc.TajoMasterProtocol;
@@ -148,9 +151,9 @@ public class TajoWorker extends CompositeService {
tajoWorkerManagerService = new TajoWorkerManagerService(workerContext, managerPort);
addService(tajoWorkerManagerService);
- LOG.info("====> Tajo worker started: mode=" + daemonMode + ", clientPort=" + clientPort + ", managerPort=" + managerPort);
+ LOG.info("Tajo worker started: mode=" + daemonMode + ", clientPort=" + clientPort + ", managerPort=" + managerPort);
} else {
- LOG.info("====> Tajo worker started: mode=" + daemonMode);
+ LOG.info("Tajo worker started: mode=" + daemonMode);
}
super.init(conf);
@@ -252,8 +255,7 @@ public class TajoWorker extends CompositeService {
}
private void setWorkerMode(String[] params) {
- if("qm".equals(daemonMode)) {
- //QueryMaster mode
+ if("qm".equals(daemonMode)) { //QueryMaster mode
String tajoMasterAddress = params[2];
connectToTajoMaster(tajoMasterAddress);
@@ -262,11 +264,9 @@ public class TajoWorker extends CompositeService {
QueryId queryId = TajoIdUtils.parseQueryId(params[1]);
tajoWorkerManagerService.getQueryMaster().reportQueryStatusToQueryMaster(
queryId, TajoProtos.QueryState.QUERY_MASTER_LAUNCHED);
- } else if("tr".equals(daemonMode)) {
- //TaskRunner mode
+ } else if("tr".equals(daemonMode)) { //TaskRunner mode
taskRunnerManager.startTask(params);
- } else {
- //Standby mode
+ } else { //Standby mode
connectToTajoMaster(tajoConf.get("tajo.master.manager.addr"));
connectToCatalog();
workerHeartbeatThread = new WorkerHeartbeatThread();
@@ -297,10 +297,10 @@ public class TajoWorker extends CompositeService {
private void connectToCatalog() {
// TODO: To be improved. it's a hack. It assumes that CatalogServer is embedded in TajoMaster.
- String hostName = this.tajoMasterAddress.getHostName();
- int port = Integer.parseInt(tajoConf.getVar(TajoConf.ConfVars.CATALOG_ADDRESS).split(":")[1]);
+ String catalogAddr = tajoConf.getVar(TajoConf.ConfVars.CATALOG_ADDRESS);
+ //int port = Integer.parseInt(tajoConf.getVar(TajoConf.ConfVars.CATALOG_ADDRESS).split(":")[1]);
try {
- catalogClient = new CatalogClient(hostName, port);
+ catalogClient = new CatalogClient(tajoConf);
} catch (IOException e) {
e.printStackTrace();
}
@@ -489,9 +489,12 @@ public class TajoWorker extends CompositeService {
String workerMode = args[0];
+ TajoConf tajoConf = new TajoConf();
+ tajoConf.addResource(new Path(TajoConstants.SYSTEM_CONF_FILENAME));
+
try {
TajoWorker tajoWorker = new TajoWorker(workerMode);
- tajoWorker.startWorker(new TajoConf(new YarnConfiguration()), args);
+ tajoWorker.startWorker(tajoConf, args);
} catch (Throwable t) {
LOG.fatal("Error starting TajoWorker", t);
System.exit(-1);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
index e1fd88a..a77ad2a 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
@@ -175,8 +175,8 @@ public class TajoWorkerClientService extends AbstractService {
builder.setSubmitTime(query.getAppSubmitTime());
builder.setInitTime(query.getInitializationTime());
builder.setHasResult(
- !(queryMasterTask.getQueryContext().getQueryMeta().isCreateTable() ||
- queryMasterTask.getQueryContext().getQueryMeta().isInsert())
+ !(queryMasterTask.getQueryTaskContext().getQueryContext().isCreateTable() ||
+ queryMasterTask.getQueryTaskContext().getQueryContext().isInsert())
);
if (query.getState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
builder.setFinishTime(query.getFinishTime());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
index 6fde6e4..28cc5f6 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
@@ -31,7 +31,7 @@ import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.TajoIdProtos;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.ipc.TajoWorkerProtocol;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.master.TaskSchedulerImpl;
import org.apache.tajo.master.event.*;
import org.apache.tajo.master.querymaster.QueryMaster;
@@ -125,13 +125,13 @@ public class TajoWorkerManagerService extends CompositeService
ExecutionBlockId ebId = new ExecutionBlockId(request.getExecutionBlockId());
QueryMasterTask queryMasterTask = workerContext.getQueryMaster().getQueryMasterTask(ebId.getQueryId());
ContainerId cid =
- queryMasterTask.getQueryContext().getResourceAllocator().makeContainerId(request.getContainerId());
+ queryMasterTask.getQueryTaskContext().getResourceAllocator().makeContainerId(request.getContainerId());
if(queryMasterTask == null || queryMasterTask.isStopped()) {
- LOG.info("====>getTask:" + cid + ", ebId:" + ebId + ", but query is finished.");
+ LOG.info("getTask:" + cid + ", ebId:" + ebId + ", but query is finished.");
done.run(TaskSchedulerImpl.stopTaskRunnerReq);
} else {
- LOG.info("====>getTask:" + cid + ", ebId:" + ebId);
+ LOG.info("getTask:" + cid + ", ebId:" + ebId);
queryMasterTask.handleTaskRequestEvent(new TaskRequestEvent(cid, ebId, done));
}
} catch (Exception e) {
@@ -202,7 +202,7 @@ public class TajoWorkerManagerService extends CompositeService
QueryId queryId = new QueryId(request.getQueryId());
LOG.info("Receive executeQuery request:" + queryId);
queryMaster.handle(new QueryStartEvent(queryId,
- new QueryMeta(request.getQueryMeta()), request.getLogicalPlanJson().getValue()));
+ new QueryContext(request.getQueryContext()), request.getLogicalPlanJson().getValue()));
done.run(TajoWorker.TRUE_PROTO);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
index aef5ead..e66751c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
@@ -30,13 +30,14 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.util.ConverterUtils;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.QueryUnitAttemptId;
+import org.apache.tajo.TajoConstants;
import org.apache.tajo.TajoProtos.TaskAttemptState;
import org.apache.tajo.TaskAttemptContext;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.statistics.TableStat;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.exception.UnfinishedTaskException;
import org.apache.tajo.engine.json.CoreGsonHelper;
import org.apache.tajo.engine.planner.PlannerUtil;
@@ -48,7 +49,7 @@ import org.apache.tajo.ipc.TajoWorkerProtocol.*;
import org.apache.tajo.ipc.TajoWorkerProtocol.TajoWorkerProtocolService.Interface;
import org.apache.tajo.ipc.protocolrecords.QueryUnitRequest;
import org.apache.tajo.master.ExecutionBlock.PartitionType;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.rpc.NullCallback;
import org.apache.tajo.storage.Fragment;
import org.apache.tajo.storage.StorageUtil;
@@ -69,8 +70,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class Task {
private static final Log LOG = LogFactory.getLog(Task.class);
- private final QueryConf conf;
- private final QueryMeta queryMeta;
+ private final TajoConf systemConf;
+ private final QueryContext queryContext;
private final FileSystem localFS;
private final TaskRunner.TaskRunnerContext taskRunnerContext;
private final Interface masterProxy;
@@ -137,8 +138,8 @@ public class Task {
this.reporter.startCommunicationThread();
this.taskId = request.getId();
- this.conf = worker.getQueryConf();
- this.queryMeta = request.getQueryMeta();
+ this.systemConf = worker.getConf();
+ this.queryContext = request.getQueryContext();
this.taskRunnerContext = worker;
this.masterProxy = masterProxy;
this.localFS = worker.getLocalFS();
@@ -146,7 +147,7 @@ public class Task {
this.taskDir = StorageUtil.concatPath(taskRunnerContext.getBaseDir(),
taskId.getQueryUnitId().getId() + "_" + taskId.getId());
- this.context = new TaskAttemptContext(conf, taskId,
+ this.context = new TaskAttemptContext(systemConf, taskId,
request.getFragments().toArray(new Fragment[request.getFragments().size()]),
taskDir);
plan = CoreGsonHelper.fromJson(request.getSerializedData(), LogicalNode.class);
@@ -163,7 +164,7 @@ public class Task {
} else {
// The final result of a task will be written in a file named part-ss-nnnnnnn,
// where ss is the subquery id associated with this task, and nnnnnn is the task id.
- Path outFilePath = new Path(queryMeta.getOutputPath(),
+ Path outFilePath = StorageUtil.concatPath(queryContext.getStagingDir(), TajoConstants.RESULT_DIR_NAME,
OUTPUT_FILE_PREFIX +
OUTPUT_FILE_FORMAT_SUBQUERY.get().format(taskId.getQueryUnitId().getExecutionBlockId().getId()) + "-" +
OUTPUT_FILE_FORMAT_TASK.get().format(taskId.getQueryUnitId().getId()));
@@ -201,7 +202,7 @@ public class Task {
if (request.getFetches().size() > 0) {
inputTableBaseDir = localFS.makeQualified(
lDirAllocator.getLocalPathForWrite(
- getTaskAttemptDir(context.getTaskId()).toString() + "/in", conf));
+ getTaskAttemptDir(context.getTaskId()).toString() + "/in", systemConf));
localFS.mkdirs(inputTableBaseDir);
Path tableDir;
for (String inputTable : context.getInputTables()) {
@@ -451,7 +452,7 @@ public class Task {
private Fragment[] localizeFetchedData(File file, String name, TableMeta meta)
throws IOException {
- Configuration c = new Configuration(conf);
+ Configuration c = new Configuration(systemConf);
c.set("fs.default.name", "file:///");
FileSystem fs = FileSystem.get(c);
Path tablePath = new Path(file.getAbsolutePath());
@@ -525,7 +526,7 @@ public class Task {
if (fetches.size() > 0) {
Path inputDir = lDirAllocator.
getLocalPathToRead(
- getTaskAttemptDir(ctx.getTaskId()).toString() + "/in", conf);
+ getTaskAttemptDir(ctx.getTaskId()).toString() + "/in", systemConf);
File storeDir;
int i = 0;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
index 6128bb3..3056595 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
@@ -32,10 +32,10 @@ import org.apache.hadoop.yarn.api.records.impl.pb.ContainerIdPBImpl;
import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.tajo.ExecutionBlockId;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.TajoProtos.TaskAttemptState;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.engine.query.QueryUnitRequestImpl;
import org.apache.tajo.ipc.TajoWorkerProtocol;
@@ -59,7 +59,7 @@ public class TaskRunner extends AbstractService {
/** class logger */
private static final Log LOG = LogFactory.getLog(TaskRunner.class);
- private QueryConf queryConf;
+ private TajoConf systemConf;
private volatile boolean stopped = false;
@@ -105,7 +105,7 @@ public class TaskRunner extends AbstractService {
private TaskRunnerManager taskRunnerManager;
- public TaskRunner(TaskRunnerManager taskRunnerManager, QueryConf conf, String[] args) {
+ public TaskRunner(TaskRunnerManager taskRunnerManager, TajoConf conf, String[] args) {
super(TaskRunner.class.getName());
this.taskRunnerManager = taskRunnerManager;
@@ -130,8 +130,7 @@ public class TaskRunner extends AbstractService {
LOG.info("QueryMaster Address:" + masterAddr);
// TODO - 'load credential' should be implemented
// Getting taskOwner
- UserGroupInformation taskOwner =
- UserGroupInformation.createRemoteUser(conf.getVar(ConfVars.QUERY_USERNAME));
+ UserGroupInformation taskOwner = UserGroupInformation.createRemoteUser(conf.getVar(ConfVars.TAJO_USERNAME));
//taskOwner.addToken(token);
// initialize MasterWorkerProtocol as an actual task owner.
@@ -161,16 +160,15 @@ public class TaskRunner extends AbstractService {
@Override
public void init(Configuration conf) {
- this.queryConf = (QueryConf)conf;
+ this.systemConf = (TajoConf)conf;
try {
// initialize DFS and LocalFileSystems
- defaultFS = FileSystem.get(URI.create(queryConf.getVar(ConfVars.ROOT_DIR)),conf);
+ defaultFS = TajoConf.getTajoRootPath(systemConf).getFileSystem(conf);
localFS = FileSystem.getLocal(conf);
// the base dir for an output dir
- baseDir = queryId.toString()
- + "/output" + "/" + executionBlockId.getId();
+ baseDir = queryId.toString() + "/output" + "/" + executionBlockId.getId();
// initialize LocalDirAllocator
lDirAllocator = new LocalDirAllocator(ConfVars.TASK_LOCAL_DIR.varname);
@@ -180,7 +178,7 @@ public class TaskRunner extends AbstractService {
// Setup QueryEngine according to the query plan
// Here, we can setup row-based query engine or columnar query engine.
- this.queryEngine = new TajoQueryEngine(queryConf);
+ this.queryEngine = new TajoQueryEngine(systemConf);
} catch (Throwable t) {
LOG.error(t);
}
@@ -222,8 +220,8 @@ public class TaskRunner extends AbstractService {
}
public class TaskRunnerContext {
- public QueryConf getQueryConf() {
- return queryConf;
+ public TajoConf getConf() {
+ return systemConf;
}
public String getNodeId() {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
index 5d8fa8d..f1ca567 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
@@ -22,7 +22,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.service.CompositeService;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.conf.TajoConf;
import java.util.HashMap;
@@ -88,13 +87,13 @@ public class TaskRunnerManager extends CompositeService {
Thread t = new Thread() {
public void run() {
try {
- QueryConf queryConf = new QueryConf(tajoConf);
- TaskRunner taskRunner = new TaskRunner(TaskRunnerManager.this, queryConf, params);
+ TajoConf systemConf = new TajoConf(tajoConf);
+ TaskRunner taskRunner = new TaskRunner(TaskRunnerManager.this, systemConf, params);
LOG.info("Start TaskRunner:" + taskRunner.getId());
synchronized(taskRunnerMap) {
taskRunnerMap.put(taskRunner.getId(), taskRunner);
}
- taskRunner.init(queryConf);
+ taskRunner.init(systemConf);
taskRunner.start();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
index 2897b14..8fc3884 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
@@ -27,7 +27,6 @@ import org.apache.hadoop.yarn.client.YarnClient;
import org.apache.hadoop.yarn.client.YarnClientImpl;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.proto.YarnProtos;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.master.TaskRunnerGroupEvent;
import org.apache.tajo.master.TaskRunnerLauncher;
@@ -47,12 +46,12 @@ public class YarnResourceAllocator extends AbstractResourceAllocator {
private static final Log LOG = LogFactory.getLog(YarnResourceAllocator.class.getName());
- private QueryMasterTask.QueryContext queryContext;
+ private QueryMasterTask.QueryMasterTaskContext queryTaskContext;
- private QueryConf queryConf;
+ private TajoConf systemConf;
- public YarnResourceAllocator(QueryMasterTask.QueryContext queryContext) {
- this.queryContext = queryContext;
+ public YarnResourceAllocator(QueryMasterTask.QueryMasterTaskContext queryTaskContext) {
+ this.queryTaskContext = queryTaskContext;
}
@Override
@@ -75,19 +74,19 @@ public class YarnResourceAllocator extends AbstractResourceAllocator {
@Override
public void init(Configuration conf) {
- queryConf = (QueryConf)conf;
+ systemConf = (TajoConf)conf;
- yarnRPC = YarnRPC.create(queryConf);
+ yarnRPC = YarnRPC.create(systemConf);
connectYarnClient();
- taskRunnerLauncher = new YarnTaskRunnerLauncherImpl(queryContext, yarnRPC);
+ taskRunnerLauncher = new YarnTaskRunnerLauncherImpl(queryTaskContext, yarnRPC);
addService((org.apache.hadoop.yarn.service.Service) taskRunnerLauncher);
- queryContext.getDispatcher().register(TaskRunnerGroupEvent.EventType.class, taskRunnerLauncher);
+ queryTaskContext.getDispatcher().register(TaskRunnerGroupEvent.EventType.class, taskRunnerLauncher);
- rmAllocator = new YarnRMContainerAllocator(queryContext);
+ rmAllocator = new YarnRMContainerAllocator(queryTaskContext);
addService(rmAllocator);
- queryContext.getDispatcher().register(ContainerAllocatorEventType.class, rmAllocator);
+ queryTaskContext.getDispatcher().register(ContainerAllocatorEventType.class, rmAllocator);
super.init(conf);
}
@@ -108,7 +107,7 @@ public class YarnResourceAllocator extends AbstractResourceAllocator {
private void connectYarnClient() {
this.yarnClient = new YarnClientImpl();
- this.yarnClient.init(queryConf);
+ this.yarnClient.init(systemConf);
this.yarnClient.start();
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
index 0694b4e..94bacdf 100644
--- a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
+++ b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
@@ -59,7 +59,7 @@ message QueryUnitRequestProto {
optional bool interQuery = 6 [default = false];
repeated Fetch fetches = 7;
optional bool shouldDie = 8;
- optional KeyValueSetProto queryMeta = 9;
+ optional KeyValueSetProto queryContext = 9;
}
message Fetch {
@@ -105,7 +105,7 @@ message Partition {
message QueryExecutionRequestProto {
required QueryIdProto queryId = 1;
- required KeyValueSetProto queryMeta = 2;
+ required KeyValueSetProto queryContext = 2;
required StringProto logicalPlanJson = 3;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/resources/tajo-default.xml
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/resources/tajo-default.xml b/tajo-core/tajo-core-backend/src/main/resources/tajo-default.xml
index 2025110..a4e39ff 100644
--- a/tajo-core/tajo-core-backend/src/main/resources/tajo-default.xml
+++ b/tajo-core/tajo-core-backend/src/main/resources/tajo-default.xml
@@ -32,6 +32,11 @@
</property>
<property>
+ <name>tajo.staging.root.dir</name>
+ <value>/tmp/tajo-${user.name}/staging</value>
+ </property>
+
+ <property>
<name>tajo.task.localdir</name>
<value>/tmp/tajo-localdir</value>
</property>
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/BackendTestingUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/BackendTestingUtil.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/BackendTestingUtil.java
index e629623..5a85a07 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/BackendTestingUtil.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/BackendTestingUtil.java
@@ -21,11 +21,9 @@
*/
package org.apache.tajo;
-import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.tajo.algebra.Expr;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.common.TajoDataTypes.Type;
@@ -33,17 +31,10 @@ import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.*;
-import org.apache.tajo.engine.planner.logical.LogicalNode;
-import org.apache.tajo.engine.planner.physical.PhysicalExec;
-import org.apache.tajo.engine.query.ResultSetImpl;
import org.apache.tajo.storage.*;
import org.apache.tajo.util.FileUtil;
-import org.apache.tajo.util.TUtil;
-import java.io.File;
import java.io.IOException;
-import java.sql.ResultSet;
-import java.util.List;
import java.util.UUID;
public class BackendTestingUtil {
@@ -102,36 +93,12 @@ public class BackendTestingUtil {
public BackendTestingUtil(TajoConf conf) throws IOException {
this.conf = conf;
- this.catalog = new LocalCatalog(conf);
+ this.catalog = new LocalCatalogWrapper(conf);
analyzer = new SQLAnalyzer();
planner = new LogicalPlanner(catalog);
optimizer = new LogicalOptimizer();
}
- public ResultSet run(String [] tableNames, File [] tables, Schema [] schemas, String query)
- throws IOException, PlanningException {
- Path workDir = createTmpTestDir();
- StorageManager sm = StorageManager.get(new TajoConf(), workDir);
- List<Fragment> frags = Lists.newArrayList();
- for (int i = 0; i < tableNames.length; i++) {
- Fragment [] splits = sm.split(tableNames[i], new Path(tables[i].getAbsolutePath()));
- for (Fragment f : splits) {
- frags.add(f);
- }
- }
-
- TaskAttemptContext ctx = new TaskAttemptContext(conf,
- TUtil.newQueryUnitAttemptId(),
- frags.toArray(new Fragment[frags.size()]), workDir);
- Expr EXPR = analyzer.parse(query);
- LogicalPlan plan = planner.createPlan(EXPR);
- LogicalNode rootNode = optimizer.optimize(plan);
- PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf, sm);
- PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
-
- return new ResultSetImpl(null, null, conf, new Path(workDir, "out"));
- }
-
public static Path createTmpTestDir() throws IOException {
String randomStr = UUID.randomUUID().toString();
FileSystem fs = FileSystem.getLocal(new Configuration());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
index 95dc212..3a70ac5 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
@@ -39,9 +39,11 @@ import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.hadoop.yarn.service.Service;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.pullserver.PullServerAuxService;
+import org.apache.tajo.util.NetUtils;
import java.io.File;
import java.io.IOException;
+import java.net.InetSocketAddress;
/**
* Configures and starts the Tajo-specific components in the YARN cluster.
@@ -64,6 +66,10 @@ public class MiniTajoYarnCluster extends MiniYARNCluster {
@Override
public void init(Configuration conf) {
+
+ conf.setSocketAddr(YarnConfiguration.RM_ADDRESS, new InetSocketAddress("127.0.0.1", 0));
+ conf.setSocketAddr(YarnConfiguration.RM_SCHEDULER_ADDRESS, new InetSocketAddress("127.0.0.1", 0));
+
conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.YARN_FRAMEWORK_NAME);
if (conf.get(MRJobConfig.MR_AM_STAGING_DIR) == null) {
conf.set(MRJobConfig.MR_AM_STAGING_DIR, new File(getTestWorkDir(),
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
index 0adf22a..88029ea 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
@@ -31,6 +31,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.client.TajoClient;
@@ -71,8 +72,7 @@ public class TajoTestingCluster {
* System property key to get test directory value.
* Name is as it is because mini dfs has hard-codings to put test data here.
*/
- public static final String TEST_DIRECTORY_KEY =
- MiniDFSCluster.PROP_TEST_BUILD_DATA;
+ public static final String TEST_DIRECTORY_KEY = MiniDFSCluster.PROP_TEST_BUILD_DATA;
/**
* Default parent directory for test output.
@@ -222,7 +222,7 @@ public class TajoTestingCluster {
}
conf.set(CatalogConstants.STORE_CLASS, "org.apache.tajo.catalog.store.MemStore");
- conf.set(CatalogConstants.JDBC_URI, "jdbc:derby:target/test-data/tcat/db");
+ conf.set(CatalogConstants.JDBC_URI, "jdbc:derby:" + clusterTestBuildDir.getAbsolutePath() + "/db");
LOG.info("Apache Derby repository is set to "+conf.get(CatalogConstants.JDBC_URI));
conf.setVar(ConfVars.CATALOG_ADDRESS, "localhost:0");
@@ -255,7 +255,7 @@ public class TajoTestingCluster {
c.setVar(ConfVars.CATALOG_ADDRESS, "localhost:0");
c.set(CatalogConstants.STORE_CLASS, "org.apache.tajo.catalog.store.MemStore");
- c.set(CatalogConstants.JDBC_URI, "jdbc:derby:target/test-data/tcat/db");
+ c.set(CatalogConstants.JDBC_URI, "jdbc:derby:" + testBuildDir.getAbsolutePath() + "/db");
LOG.info("derby repository is set to "+conf.get(CatalogConstants.JDBC_URI));
@@ -300,7 +300,7 @@ public class TajoTestingCluster {
tajoWorker.startWorker(workerConf, new String[]{"standby"});
- LOG.info("=====> MiniTajoCluster Worker #" + (i + 1) + " started.");
+ LOG.info("MiniTajoCluster Worker #" + (i + 1) + " started.");
tajoWorkers.add(tajoWorker);
}
}
@@ -348,14 +348,10 @@ public class TajoTestingCluster {
*/
public void startMiniCluster(final int numSlaves)
throws Exception {
- String localHostName = InetAddress.getLocalHost().getHostName();
- startMiniCluster(numSlaves, new String[] {localHostName});
+ startMiniCluster(numSlaves, null);
}
- public void startMiniCluster(final int numSlaves,
- final String [] dataNodeHosts) throws Exception {
- // the conf is set to the distributed mode.
- this.conf.setBoolVar(ConfVars.CLUSTER_DISTRIBUTED, true);
+ public void startMiniCluster(final int numSlaves, final String [] dataNodeHosts) throws Exception {
int numDataNodes = numSlaves;
if(dataNodeHosts != null && dataNodeHosts.length != 0) {
@@ -406,12 +402,11 @@ public class TajoTestingCluster {
yarnCluster.init(conf);
yarnCluster.start();
- conf.set(YarnConfiguration.RM_ADDRESS,
- NetUtils.normalizeInetSocketAddress(yarnCluster.getResourceManager().
- getClientRMService().getBindAddress()));
- conf.set(YarnConfiguration.RM_SCHEDULER_ADDRESS,
- NetUtils.normalizeInetSocketAddress(yarnCluster.getResourceManager().
- getApplicationMasterService().getBindAddress()));
+ ResourceManager resourceManager = yarnCluster.getResourceManager();
+ InetSocketAddress rmAddr = resourceManager.getClientRMService().getBindAddress();
+ InetSocketAddress rmSchedulerAddr = resourceManager.getApplicationMasterService().getBindAddress();
+ conf.set(YarnConfiguration.RM_ADDRESS, NetUtils.normalizeInetSocketAddress(rmAddr));
+ conf.set(YarnConfiguration.RM_SCHEDULER_ADDRESS, NetUtils.normalizeInetSocketAddress(rmSchedulerAddr));
URL url = Thread.currentThread().getContextClassLoader().getResource("yarn-site.xml");
if (url == null) {
@@ -425,9 +420,6 @@ public class TajoTestingCluster {
}
public void startMiniClusterInLocal(final int numSlaves) throws Exception {
- // the conf is set to the distributed mode.
- this.conf.setBoolVar(ConfVars.CLUSTER_DISTRIBUTED, true);
-
// If we already put up a cluster, fail.
String testBuildPath = conf.get(TEST_DIRECTORY_KEY, null);
isRunningCluster(testBuildPath);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java
index a16d0f3..ba7d36b 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java
@@ -22,15 +22,13 @@
package org.apache.tajo.engine.query;
import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.*;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.tajo.IntegrationTest;
import org.apache.tajo.TajoTestingCluster;
-import org.apache.tajo.catalog.CatalogUtil;
-import org.apache.tajo.catalog.Schema;
-import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.catalog.statistics.TableStat;
import org.apache.tajo.common.TajoDataTypes.Type;
@@ -52,6 +50,7 @@ public class TestResultSetImpl {
private static TajoTestingCluster util;
private static TajoConf conf;
private static StorageManager sm;
+ private static TableDesc desc;
private static TableMeta scoreMeta;
@BeforeClass
@@ -90,7 +89,7 @@ public class TestResultSetImpl {
stat.setNumBlocks(1000);
stat.setNumPartitions(100);
scoreMeta.setStat(stat);
- sm.writeTableMeta(sm.getTablePath("score"), scoreMeta);
+ desc = new TableDescImpl("score", scoreMeta, p);
}
@AfterClass
@@ -100,7 +99,7 @@ public class TestResultSetImpl {
@Test
public void test() throws IOException, SQLException {
- ResultSetImpl rs = new ResultSetImpl(null, null, conf, sm.getTablePath("score"));
+ ResultSetImpl rs = new ResultSetImpl(null, null, conf, desc);
ResultSetMetaData meta = rs.getMetaData();
assertNotNull(meta);
Schema schema = scoreMeta.getSchema();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/test/resources/tajo-default.xml
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/resources/tajo-default.xml b/tajo-core/tajo-core-backend/src/test/resources/tajo-default.xml
index aedaac3..7012cd9 100644
--- a/tajo-core/tajo-core-backend/src/test/resources/tajo-default.xml
+++ b/tajo-core/tajo-core-backend/src/test/resources/tajo-default.xml
@@ -32,6 +32,11 @@
</property>
<property>
+ <name>tajo.staging.root.dir</name>
+ <value>/tmp/tajo-${user.name}/staging</value>
+ </property>
+
+ <property>
<name>tajo.task.localdir</name>
<value>/tmp/tajo-localdir</value>
</property>
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-pullserver/src/main/resources/tajo-default.xml
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-pullserver/src/main/resources/tajo-default.xml b/tajo-core/tajo-core-pullserver/src/main/resources/tajo-default.xml
deleted file mode 100644
index 72f0bb9..0000000
--- a/tajo-core/tajo-core-pullserver/src/main/resources/tajo-default.xml
+++ /dev/null
@@ -1,33 +0,0 @@
-<?xml version="1.0"?>
-<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
-
-<!--
- Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
- -->
-
-<configuration>
- <property>
- <name>tajo.cluster.distributed</name>
- <value>false</value>
- </property>
-
- <property>
- <name>tajo.rootdir</name>
- <value>file:///tmp/tajo-${user.name}</value>
- <description>A base for other temporary directories.</description>
- </property>
-</configuration>
[3/3] git commit: TAJO-149: Eliminate QueryConf and its file write.
(hyunsik)
Posted by hy...@apache.org.
TAJO-149: Eliminate QueryConf and its file write. (hyunsik)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/27f76811
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/27f76811
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/27f76811
Branch: refs/heads/master
Commit: 27f76811c4dc279cd2887adf900466c3dba402b3
Parents: e448b5e
Author: Hyunsik Choi <hy...@apache.org>
Authored: Mon Sep 9 10:55:33 2013 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Mon Sep 9 11:23:14 2013 +0900
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../org/apache/tajo/catalog/CatalogClient.java | 3 +-
.../org/apache/tajo/catalog/CatalogServer.java | 24 +--
.../org/apache/tajo/catalog/LocalCatalog.java | 53 ------
.../tajo/catalog/LocalCatalogWrapper.java | 53 ++++++
.../apache/tajo/catalog/MiniCatalogServer.java | 2 +-
.../org/apache/tajo/catalog/TestCatalog.java | 9 +-
.../src/test/resources/catalog-default.xml | 2 +-
.../java/org/apache/tajo/TajoConstants.java | 9 +-
.../java/org/apache/tajo/conf/TajoConf.java | 107 +++++++----
.../main/java/org/apache/tajo/QueryConf.java | 38 ----
.../java/org/apache/tajo/client/TajoClient.java | 2 +-
.../engine/planner/logical/StoreTableNode.java | 8 -
.../planner/physical/IndexedStoreExec.java | 6 +-
.../tajo/engine/query/QueryUnitRequestImpl.java | 32 ++--
.../apache/tajo/engine/query/ResultSetImpl.java | 42 ++---
.../ipc/protocolrecords/QueryUnitRequest.java | 4 +-
.../org/apache/tajo/master/ContainerProxy.java | 4 +-
.../org/apache/tajo/master/GlobalEngine.java | 51 +++---
.../org/apache/tajo/master/QueryContext.java | 171 +++++++++++++++++
.../java/org/apache/tajo/master/QueryMeta.java | 118 ------------
.../apache/tajo/master/TajoContainerProxy.java | 12 +-
.../java/org/apache/tajo/master/TajoMaster.java | 180 ++++++++++++------
.../apache/tajo/master/TaskSchedulerImpl.java | 12 +-
.../apache/tajo/master/YarnContainerProxy.java | 54 ++----
.../tajo/master/YarnTaskRunnerLauncherImpl.java | 4 +-
.../tajo/master/event/QueryStartEvent.java | 12 +-
.../apache/tajo/master/querymaster/Query.java | 142 ++++++++++-----
.../master/querymaster/QueryInProgress.java | 12 +-
.../master/querymaster/QueryJobManager.java | 6 +-
.../tajo/master/querymaster/QueryMaster.java | 24 ++-
.../master/querymaster/QueryMasterRunner.java | 18 +-
.../master/querymaster/QueryMasterTask.java | 182 +++++++++----------
.../tajo/master/querymaster/SubQuery.java | 17 +-
.../master/rm/TajoWorkerResourceManager.java | 11 +-
.../master/rm/YarnRMContainerAllocator.java | 4 +-
.../tajo/master/rm/YarnTajoResourceManager.java | 1 -
.../tajo/worker/TajoResourceAllocator.java | 62 ++-----
.../java/org/apache/tajo/worker/TajoWorker.java | 27 +--
.../tajo/worker/TajoWorkerClientService.java | 4 +-
.../tajo/worker/TajoWorkerManagerService.java | 10 +-
.../main/java/org/apache/tajo/worker/Task.java | 23 +--
.../java/org/apache/tajo/worker/TaskRunner.java | 22 +--
.../apache/tajo/worker/TaskRunnerManager.java | 7 +-
.../tajo/worker/YarnResourceAllocator.java | 23 ++-
.../src/main/proto/TajoWorkerProtocol.proto | 4 +-
.../src/main/resources/tajo-default.xml | 5 +
.../org/apache/tajo/BackendTestingUtil.java | 35 +---
.../org/apache/tajo/MiniTajoYarnCluster.java | 6 +
.../org/apache/tajo/TajoTestingCluster.java | 32 ++--
.../tajo/engine/query/TestResultSetImpl.java | 9 +-
.../src/test/resources/tajo-default.xml | 5 +
.../src/main/resources/tajo-default.xml | 33 ----
.../org/apache/tajo/storage/StorageManager.java | 14 +-
54 files changed, 885 insertions(+), 867 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e5df10b..9fb16b4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -23,6 +23,8 @@ Release 0.2.0 - unreleased
IMPROVEMENTS
+ TAJO-149: Eliminate QueryConf and its file write. (hyunsik)
+
TAJO-143: Implement hash semi-join operator. (hyunsik)
TAJO-142: Implement hash anti-join operator. (hyunsik)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java b/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
index 7977b58..5190998 100644
--- a/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
+++ b/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
@@ -56,9 +56,8 @@ public class CatalogClient extends AbstractCatalogClient {
client = new ProtoBlockingRpcClient(CatalogProtocol.class, serverAddr);
setStub((BlockingInterface) client.getStub());
} catch (Exception e) {
- throw new IOException(e);
+ throw new IOException("Can't connect the catalog server (" + addrStr +")");
}
-
LOG.info("Connected to the catalog server (" + addrStr + ")");
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
index f0a0584..3ac4ed0 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
@@ -66,7 +66,7 @@ public class CatalogServer extends AbstractService {
// RPC variables
private ProtoBlockingRpcServer rpcServer;
private InetSocketAddress bindAddress;
- private String serverName;
+ private String bindAddressStr;
final CatalogProtocolHandler handler;
// Server status variables
@@ -94,14 +94,14 @@ public class CatalogServer extends AbstractService {
@Override
public void init(Configuration _conf) {
- this.conf = (TajoConf) _conf;
+ this.conf = (TajoConf)_conf;
Constructor<?> cons;
try {
Class<?> storeClass =
this.conf.getClass(CatalogConstants.STORE_CLASS, DBStore.class);
- LOG.info("Catalog Store Class: " + storeClass.getCanonicalName());
+ LOG.info("Catalog Store Class: " + storeClass.getCanonicalName());
cons = storeClass.
getConstructor(new Class [] {Configuration.class});
@@ -109,7 +109,8 @@ public class CatalogServer extends AbstractService {
initBuiltinFunctions(builtingFuncs);
} catch (Throwable t) {
- LOG.error("cannot initialize CatalogServer", t);
+ LOG.error("CatalogServer initialization failed", t);
+ throw new CatalogException(t);
}
super.init(conf);
@@ -123,9 +124,7 @@ public class CatalogServer extends AbstractService {
}
public void start() {
- // Server to handle client requests.
String serverAddr = conf.getVar(ConfVars.CATALOG_ADDRESS);
- // Creation of a HSA will force a resolve.
InetSocketAddress initIsa = NetUtils.createSocketAddr(serverAddr);
try {
this.rpcServer = new ProtoBlockingRpcServer(
@@ -134,13 +133,14 @@ public class CatalogServer extends AbstractService {
this.rpcServer.start();
this.bindAddress = NetUtils.getConnectAddress(this.rpcServer.getListenAddress());
- this.serverName = NetUtils.normalizeInetSocketAddress(bindAddress);
- conf.setVar(ConfVars.CATALOG_ADDRESS, serverName);
+ this.bindAddressStr = NetUtils.normalizeInetSocketAddress(bindAddress);
+ conf.set(ConfVars.CATALOG_ADDRESS.varname, bindAddressStr);
} catch (Exception e) {
- LOG.error("Cannot start RPC Server of CatalogServer", e);
+ LOG.error("CatalogServer startup failed", e);
+ throw new CatalogException(e);
}
- LOG.info("Catalog Server startup (" + serverName + ")");
+ LOG.info("Catalog Server startup (" + bindAddressStr + ")");
super.start();
}
@@ -148,7 +148,7 @@ public class CatalogServer extends AbstractService {
if (rpcServer != null) {
this.rpcServer.shutdown();
}
- LOG.info("Catalog Server (" + serverName + ") shutdown");
+ LOG.info("Catalog Server (" + bindAddressStr + ") shutdown");
super.stop();
}
@@ -245,7 +245,7 @@ public class CatalogServer extends AbstractService {
} finally {
wlock.unlock();
LOG.info("Table " + tableDesc.getId() + " is added to the catalog ("
- + serverName + ")");
+ + bindAddressStr + ")");
}
return BOOL_TRUE;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalog.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalog.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalog.java
deleted file mode 100644
index 3a98ca0..0000000
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalog.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- *
- */
-package org.apache.tajo.catalog;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.tajo.conf.TajoConf;
-
-import java.io.IOException;
-
-/**
- * This class provides a catalog service interface in
- * local.
- */
-public class LocalCatalog extends AbstractCatalogClient {
- private static final Log LOG = LogFactory.getLog(LocalCatalog.class);
- private CatalogServer catalog;
-
- public LocalCatalog(final TajoConf conf) throws IOException {
- this.catalog = new CatalogServer();
- this.catalog.init(conf);
- this.catalog.start();
- setStub(catalog.getHandler());
- }
-
- public LocalCatalog(final CatalogServer server) {
- this.catalog = server;
- setStub(server.getHandler());
- }
-
- public void shutdown() {
- this.catalog.stop();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalogWrapper.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalogWrapper.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalogWrapper.java
new file mode 100644
index 0000000..13cbaa1
--- /dev/null
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/LocalCatalogWrapper.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ *
+ */
+package org.apache.tajo.catalog;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.conf.TajoConf;
+
+import java.io.IOException;
+
+/**
+ * This class provides a catalog service interface in
+ * local.
+ */
+public class LocalCatalogWrapper extends AbstractCatalogClient {
+ private static final Log LOG = LogFactory.getLog(LocalCatalogWrapper.class);
+ private CatalogServer catalog;
+
+ public LocalCatalogWrapper(final TajoConf conf) throws IOException {
+ this.catalog = new CatalogServer();
+ this.catalog.init(conf);
+ this.catalog.start();
+ setStub(catalog.getHandler());
+ }
+
+ public LocalCatalogWrapper(final CatalogServer server) {
+ this.catalog = server;
+ setStub(server.getHandler());
+ }
+
+ public void shutdown() {
+ this.catalog.stop();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
index 79f0af9..70ed314 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
@@ -44,6 +44,6 @@ public class MiniCatalogServer {
}
public CatalogService getCatalog() {
- return new LocalCatalog(this.catalogServers);
+ return new LocalCatalogWrapper(this.catalogServers);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
index 5ac59d2..2bf944d 100644
--- a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
+++ b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
@@ -48,10 +48,15 @@ public class TestCatalog {
@BeforeClass
public static void setUp() throws Exception {
+ TajoConf conf = new TajoConf();
+
+ conf.set(CatalogConstants.JDBC_URI, "jdbc:derby:target/test-data/TestCatalog/db");
+ conf.setVar(TajoConf.ConfVars.CATALOG_ADDRESS, "127.0.0.1:0");
+
server = new CatalogServer();
- server.init(new TajoConf());
+ server.init(conf);
server.start();
- catalog = new LocalCatalog(server);
+ catalog = new LocalCatalogWrapper(server);
}
@AfterClass
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-catalog/tajo-catalog-server/src/test/resources/catalog-default.xml
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/test/resources/catalog-default.xml b/tajo-catalog/tajo-catalog-server/src/test/resources/catalog-default.xml
index 81384b7..5349618 100644
--- a/tajo-catalog/tajo-catalog-server/src/test/resources/catalog-default.xml
+++ b/tajo-catalog/tajo-catalog-server/src/test/resources/catalog-default.xml
@@ -22,7 +22,7 @@
<configuration>
<property>
<name>tajo.catalog.master.addr</name>
- <value>127.0.0.1:0</value>
+ <value>0.0.0.0:0</value>
</property>
<property>
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java b/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
index cfef181..c84b8de 100644
--- a/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
+++ b/tajo-common/src/main/java/org/apache/tajo/TajoConstants.java
@@ -19,10 +19,13 @@
package org.apache.tajo;
public class TajoConstants {
+ public static String SYSTEM_CONF_FILENAME = "system_conf.xml";
- public static String WAREHOUSE_DIR = "warehouse";
-
- public static String USER_QUERYDIR_PREFIX = "tajo";
+ public static String SYSTEM_DIR_NAME = "system";
+ public static String WAREHOUSE_DIR_NAME = "warehouse";
+ public static String STAGING_DIR_NAME = "staging";
+ public static String SYSTEM_RESOURCE_DIR_NAME = "resource";
+ public static String RESULT_DIR_NAME="RESULT";
private TajoConstants() {}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 2261f6c..131990e 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -18,16 +18,19 @@
package org.apache.tajo.conf;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.tajo.TajoConstants;
import java.io.PrintStream;
import java.util.Map;
public class TajoConf extends YarnConfiguration {
- static{
+
+ static {
Configuration.addDefaultResource("catalog-default.xml");
Configuration.addDefaultResource("catalog-site.xml");
Configuration.addDefaultResource("storage-default.xml");
@@ -36,6 +39,8 @@ public class TajoConf extends YarnConfiguration {
Configuration.addDefaultResource("tajo-site.xml");
}
+ private static final String EMPTY_VALUE = "";
+
private static final Map<String, ConfVars> vars = Maps.newHashMap();
public TajoConf() {
@@ -53,38 +58,36 @@ public class TajoConf extends YarnConfiguration {
public static enum ConfVars {
//////////////////////////////////
- // System Configuration
+ // Tajo System Configuration
//////////////////////////////////
- // Cluster Mode
- CLUSTER_DISTRIBUTED("tajo.cluster.distributed", false),
+ // a username for a running Tajo cluster
+ TAJO_USERNAME("tajo.cluster.username", "tajo"),
- // System Directories
+ // Configurable System Directories
ROOT_DIR("tajo.rootdir", "/tajo"),
- TASK_LOCAL_DIR("tajo.task.localdir", ""),
+ WAREHOUSE_DIR("tajo.wahrehouse-dir", EMPTY_VALUE),
+ STAGING_ROOT_DIR("tajo.staging.root.dir", ""),
+ TASK_LOCAL_DIR("tajo.task.localdir", ""), // local directory for temporal files
+ SYSTEM_CONF_PATH("tajo.system.conf.path", ""),
+ SYSTEM_CONF_REPLICA_COUNT("tajo.system.conf.replica.count", 20),
// Service Addresses
- TASKRUNNER_LISTENER_ADDRESS("tajo.master.taskrunnerlistener.addr", "0.0.0.0:0"), // used internally
+ TASKRUNNER_LISTENER_ADDRESS("tajo.master.taskrunnerlistener.addr", "0.0.0.0:0"),
CLIENT_SERVICE_ADDRESS("tajo.master.clientservice.addr", "127.0.0.1:9004"),
TAJO_MASTER_SERVICE_ADDRESS("tajo.master.manager.addr", "0.0.0.0:9005"),
- //////////////////////////////////
- // Catalog Configuration
- //////////////////////////////////
+ // Resource Manager
+ RESOURCE_MANAGER_CLASS("tajo.resource.manager", "org.apache.tajo.master.rm.YarnTajoResourceManager"),
+
+ // Catalog Address
CATALOG_ADDRESS("tajo.catalog.master.addr", "0.0.0.0:9002"),
//////////////////////////////////
- // AM and RM
+ // Worker
//////////////////////////////////
- AM_TO_RM_HEARTBEAT_INTERVAL_MS("tajo.am.scheduler.heartbeat.interval-ms", 1000),
- AM_TO_RM_WAIT_INTERVAL_MS("tajo.am.scheduler.connection.wait.interval-ms", 360000L),
- /** Enable blacklisting of nodes in the job.*/
- AM_QUERY_NODE_BLACKLISTING_ENABLE("tajo.query.node-blacklisting.enable", true),
- MAX_TASK_FAILURES_PER_TRACKER("tajo.query.maxtaskfailures.per.worker", 3),
- AM_IGNORE_BLACKLISTING_BLACKLISTED_NODE_PERECENT("tajo.query.node-blacklisting.ignore-threshold-node-percent", 33),
/** how many launching TaskRunners in parallel */
AM_TASKRUNNER_LAUNCH_PARALLEL_NUM("tajo.master.taskrunnerlauncher.parallel.num", 16),
-
MAX_WORKER_PER_NODE("tajo.query.max-workernum.per.node", 8),
//////////////////////////////////
@@ -123,31 +126,10 @@ public class TajoConf extends YarnConfiguration {
//////////////////////////////////
// The Below is reserved
//////////////////////////////////
- WORKING_DIR("tajo.query.workingdir", null),
-
- // Query Master
- QUERY_AM_VMEM_MB("tajo.query.am.vmem", 4096),
- QUERY_AM_JAVA_OPT("tajo.query.am.javaopt", "-Xmx1024m"),
-
- // QueryC
- QUERY_NAME("tajo.query.name", "tajo query"),
- QUEUE_NAME("tajo.query.queue.name", "default"),
- QUERY_USERNAME("tajo.query.user.name", "anonymous"),
- QUERY_OUTPUT_DIR_SUCCESSFUL_MARKER("tajo.query.output.successfulmaker", true),
- QUERY_OUTPUT_DIR("tajo.query.output.dir", ""),
- QUERY_OUTPUT_TABLE("tajo.query.output.table", ""),
-
- // Task
- LOCAL_TMP_DIR("tajo.task.local.tmpdir", null),
-
- // TaskAttempt
- APPLICATION_ATTEMPT_ID("tajo.app.attempt.id", 0),
- TASK_ATTEMPT_ID("tajo.task.attempt.id", ""),
- ATTEMPT_ID("tajo.query.attempt.id", ""),
- FINAL_SUBQUERY_ID("tajo.query.finalsubquery.id", ""),
// GeoIP
GEOIP_DATA("tajo.geoip.data", "/usr/local/share/GeoIP/GeoIP.dat"),
+
;
public final String varname;
@@ -349,4 +331,49 @@ public class TajoConf extends YarnConfiguration {
ps.println(one.varname + "=" + ((get(one.varname) != null) ? get(one.varname) : ""));
}
}
+
+ /////////////////////////////////////////////////////////////////////////////
+ // Tajo System Specific Methods
+ /////////////////////////////////////////////////////////////////////////////
+
+ public static Path getTajoRootPath(TajoConf conf) {
+ String rootPath = conf.getVar(ConfVars.ROOT_DIR);
+ Preconditions.checkNotNull(rootPath,
+ ConfVars.ROOT_DIR.varname + " must be set before a Tajo Cluster starts up");
+ return new Path(rootPath);
+ }
+
+ public static Path getWarehousePath(TajoConf conf) {
+ String warehousePath = conf.getVar(ConfVars.WAREHOUSE_DIR);
+ if (warehousePath == null || warehousePath.equals("")) {
+ Path rootDir = getTajoRootPath(conf);
+ warehousePath = new Path(rootDir, TajoConstants.WAREHOUSE_DIR_NAME).toString();
+ conf.setVar(ConfVars.WAREHOUSE_DIR, warehousePath);
+ return new Path(warehousePath);
+ } else {
+ return new Path(warehousePath);
+ }
+ }
+
+ public static Path getSystemPath(TajoConf conf) {
+ Path rootPath = getTajoRootPath(conf);
+ return new Path(rootPath, TajoConstants.SYSTEM_DIR_NAME);
+ }
+
+ public static Path getSystemResourcePath(TajoConf conf) {
+ return new Path(getSystemPath(conf), TajoConstants.SYSTEM_RESOURCE_DIR_NAME);
+ }
+
+ public static Path getStagingRoot(TajoConf conf) {
+ String stagingRootDir = conf.getVar(ConfVars.STAGING_ROOT_DIR);
+ Preconditions.checkState(stagingRootDir != null && !stagingRootDir.equals(""),
+ TajoConstants.STAGING_DIR_NAME + " must be set before starting a Tajo Cluster starts up");
+ return new Path(stagingRootDir);
+ }
+
+ public static Path getSystemConf(TajoConf conf) {
+ String stagingRootDir = conf.getVar(ConfVars.SYSTEM_CONF_PATH);
+ Preconditions.checkNotNull(stagingRootDir, ConfVars.SYSTEM_CONF_PATH.varname + " is not set.");
+ return new Path(stagingRootDir);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java
deleted file mode 100644
index f2356a5..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.tajo.conf.TajoConf;
-
-public class QueryConf extends TajoConf {
- public static String FILENAME = "queryconf.xml";
- public static String QUERY_MASTER_FILENAME = "querymasterconf.xml";
-
- public QueryConf() {
- super();
- }
-
- public QueryConf(Configuration conf) {
- super(conf);
- if (! (conf instanceof QueryConf)) {
- this.reloadConfiguration();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
index aa67376..068ece5 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
@@ -211,7 +211,7 @@ public class TajoClient {
}
TableDesc tableDesc = getResultDesc(queryId);
- return new ResultSetImpl(this, queryId, conf, tableDesc.getPath());
+ return new ResultSetImpl(this, queryId, conf, tableDesc);
}
public ResultSet getQueryResultAndWait(QueryId queryId)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
index f4b63c7..817ca35 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
@@ -120,14 +120,6 @@ public class StoreTableNode extends UnaryNode implements Cloneable {
public void setCreateTable() {
isCreatedTable = true;
}
-
- public boolean isOverwrite() {
- return isOverwritten;
- }
-
- public void setOverwrite() {
- isOverwritten = true;
- }
@Override
public boolean equals(Object obj) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
index 03eeb51..dc7f08c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/IndexedStoreExec.java
@@ -85,6 +85,7 @@ public class IndexedStoreExec extends UnaryPhysicalExec {
public Tuple next() throws IOException {
Tuple tuple;
Tuple keyTuple;
+ Tuple prevKeyTuple = null;
long offset;
@@ -93,7 +94,10 @@ public class IndexedStoreExec extends UnaryPhysicalExec {
appender.addTuple(tuple);
keyTuple = new VTuple(keySchema.getColumnNum());
RowStoreUtil.project(tuple, keyTuple, indexKeys);
- indexWriter.write(keyTuple, offset);
+ if (prevKeyTuple == null || !prevKeyTuple.equals(keyTuple)) {
+ indexWriter.write(keyTuple, offset);
+ prevKeyTuple = keyTuple;
+ }
}
return null;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
index 3697984..a0422af 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
@@ -23,7 +23,7 @@ import org.apache.tajo.ipc.TajoWorkerProtocol.Fetch;
import org.apache.tajo.ipc.TajoWorkerProtocol.QueryUnitRequestProto;
import org.apache.tajo.ipc.TajoWorkerProtocol.QueryUnitRequestProtoOrBuilder;
import org.apache.tajo.ipc.protocolrecords.QueryUnitRequest;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.storage.Fragment;
import java.net.URI;
@@ -41,7 +41,7 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
private Boolean interQuery;
private List<Fetch> fetches;
private Boolean shouldDie;
- private QueryMeta queryMeta;
+ private QueryContext queryContext;
private QueryUnitRequestProto proto = QueryUnitRequestProto.getDefaultInstance();
private QueryUnitRequestProto.Builder builder = null;
@@ -55,9 +55,9 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
public QueryUnitRequestImpl(QueryUnitAttemptId id, List<Fragment> fragments,
String outputTable, boolean clusteredOutput,
- String serializedData, QueryMeta queryMeta) {
+ String serializedData, QueryContext queryContext) {
this();
- this.set(id, fragments, outputTable, clusteredOutput, serializedData, queryMeta);
+ this.set(id, fragments, outputTable, clusteredOutput, serializedData, queryContext);
}
public QueryUnitRequestImpl(QueryUnitRequestProto proto) {
@@ -69,14 +69,14 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
public void set(QueryUnitAttemptId id, List<Fragment> fragments,
String outputTable, boolean clusteredOutput,
- String serializedData, QueryMeta queryMeta) {
+ String serializedData, QueryContext queryContext) {
this.id = id;
this.fragments = fragments;
this.outputTable = outputTable;
this.clusteredOutput = clusteredOutput;
this.serializedData = serializedData;
this.isUpdated = true;
- this.queryMeta = queryMeta;
+ this.queryContext = queryContext;
}
@Override
@@ -182,21 +182,21 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
}
- public QueryMeta getQueryMeta() {
+ public QueryContext getQueryContext() {
QueryUnitRequestProtoOrBuilder p = viaProto ? proto : builder;
- if (queryMeta != null) {
- return queryMeta;
+ if (queryContext != null) {
+ return queryContext;
}
- if (!p.hasQueryMeta()) {
+ if (!p.hasQueryContext()) {
return null;
}
- this.queryMeta = new QueryMeta(p.getQueryMeta());
- return this.queryMeta;
+ this.queryContext = new QueryContext(p.getQueryContext());
+ return this.queryContext;
}
- public void setQueryMeta(QueryMeta queryMeta) {
+ public void setQueryContext(QueryContext queryContext) {
maybeInitBuilder();
- this.queryMeta = queryMeta;
+ this.queryContext = queryContext;
}
public List<Fetch> getFetches() {
@@ -269,8 +269,8 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
if (this.shouldDie != null) {
builder.setShouldDie(this.shouldDie);
}
- if (this.queryMeta != null) {
- builder.setQueryMeta(queryMeta.getProto());
+ if (this.queryContext != null) {
+ builder.setQueryContext(queryContext.getProto());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/ResultSetImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/ResultSetImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/ResultSetImpl.java
index 22186e2..6b49558 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/ResultSetImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/ResultSetImpl.java
@@ -22,6 +22,8 @@ import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.tajo.QueryId;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.TableMetaImpl;
import org.apache.tajo.catalog.proto.CatalogProtos.TableProto;
@@ -50,7 +52,8 @@ public class ResultSetImpl implements ResultSet {
private Configuration conf;
private FileSystem fs;
private Scanner scanner;
- private TableMeta meta;
+ private TableDesc desc;
+ private Schema schema;
private Tuple cur;
private int curRow;
private long totalRow;
@@ -58,32 +61,23 @@ public class ResultSetImpl implements ResultSet {
private TajoClient tajoClient;
QueryId queryId;
- public ResultSetImpl(TajoClient tajoClient, QueryId queryId) throws IOException {
- this(tajoClient, queryId, null, null);
+ public ResultSetImpl(TajoClient tajoClient, QueryId queryId) {
+ this.tajoClient = tajoClient;
+ this.queryId = queryId;
+ init();
}
-// public ResultSetImpl(TajoClient tajoClient, QueryId queryId, Configuration conf, String path) throws IOException {
-// this(tajoClient, queryId, conf, new Path(path));
-// }
-
- public ResultSetImpl(TajoClient tajoClient, QueryId queryId, Configuration conf, Path path) throws IOException {
+ public ResultSetImpl(TajoClient tajoClient, QueryId queryId, Configuration conf, TableDesc desc) throws IOException {
this.tajoClient = tajoClient;
this.queryId = queryId;
this.conf = conf;
- if(path != null) {
- this.fs = path.getFileSystem(this.conf);
- // TODO - to be improved. It can be solved to get the query finish status
- // from master.
- try {
- this.meta = getMeta(this.conf, path);
- } catch (FileNotFoundException fnf) {
- this.totalRow = 0;
- init();
- return;
- }
- this.totalRow = meta.getStat() != null ? meta.getStat().getNumRows() : 0;
- Collection<Fragment> frags = getFragmentsNG(meta, path);
- scanner = new MergeScanner(conf, meta, frags);
+ this.desc = desc;
+ this.schema = desc.getMeta().getSchema();
+ if(desc != null) {
+ fs = desc.getPath().getFileSystem(conf);
+ this.totalRow = desc.getMeta().getStat() != null ? desc.getMeta().getStat().getNumRows() : 0;
+ Collection<Fragment> frags = getFragmentsNG(desc.getMeta(), desc.getPath());
+ scanner = new MergeScanner(conf, desc.getMeta(), frags);
}
init();
}
@@ -273,7 +267,7 @@ public class ResultSetImpl implements ResultSet {
*/
@Override
public int findColumn(String colName) throws SQLException {
- return this.meta.getSchema().getColumnIdByName(colName);
+ return schema.getColumnIdByName(colName);
}
/*
@@ -714,7 +708,7 @@ public class ResultSetImpl implements ResultSet {
*/
@Override
public ResultSetMetaData getMetaData() throws SQLException {
- return new ResultSetMetaDataImpl(meta);
+ return new ResultSetMetaDataImpl(desc.getMeta());
}
/*
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
index ead5e73..0775c3c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
@@ -24,7 +24,7 @@ package org.apache.tajo.ipc.protocolrecords;
import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.common.ProtoObject;
import org.apache.tajo.ipc.TajoWorkerProtocol;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
import org.apache.tajo.storage.Fragment;
import java.net.URI;
@@ -43,5 +43,5 @@ public interface QueryUnitRequest extends ProtoObject<TajoWorkerProtocol.QueryUn
public List<TajoWorkerProtocol.Fetch> getFetches();
public boolean shouldDie();
public void setShouldDie();
- public QueryMeta getQueryMeta();
+ public QueryContext getQueryContext();
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/ContainerProxy.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/ContainerProxy.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/ContainerProxy.java
index 044f0ae..59b071a 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/ContainerProxy.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/ContainerProxy.java
@@ -41,7 +41,7 @@ public abstract class ContainerProxy {
protected final ExecutionBlockId executionBlockId;
protected Configuration conf;
- protected QueryMasterTask.QueryContext context;
+ protected QueryMasterTask.QueryMasterTaskContext context;
protected ContainerState state;
// store enough information to be able to cleanup the container
@@ -53,7 +53,7 @@ public abstract class ContainerProxy {
public abstract void launch(ContainerLaunchContext containerLaunchContext);
public abstract void stopContainer();
- public ContainerProxy(QueryMasterTask.QueryContext context, Configuration conf,
+ public ContainerProxy(QueryMasterTask.QueryMasterTaskContext context, Configuration conf,
ExecutionBlockId executionBlockId, Container container) {
this.context = context;
this.conf = conf;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
index 4e5e5ce..2cd81a8 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -33,6 +33,7 @@ import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.exception.AlreadyExistsTableException;
import org.apache.tajo.catalog.exception.NoSuchTableException;
import org.apache.tajo.catalog.statistics.TableStat;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.NullDatum;
import org.apache.tajo.engine.eval.ConstEval;
import org.apache.tajo.engine.eval.FieldEval;
@@ -117,13 +118,13 @@ public class GlobalEngine extends AbstractService {
responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
responseBuilder.setState(TajoProtos.QueryState.QUERY_SUCCEEDED);
} else {
- QueryMeta queryMeta = new QueryMeta();
- hookManager.doHooks(queryMeta, plan);
+ QueryContext queryContext = new QueryContext();
+ hookManager.doHooks(queryContext, plan);
QueryJobManager queryJobManager = this.context.getQueryJobManager();
QueryInfo queryInfo;
- queryInfo = queryJobManager.createNewQueryJob(queryMeta, sql, rootNode);
+ queryInfo = queryJobManager.createNewQueryJob(queryContext, sql, rootNode);
responseBuilder.setQueryId(queryInfo.getQueryId().getProto());
responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
@@ -274,8 +275,8 @@ public class GlobalEngine extends AbstractService {
}
public interface DistributedQueryHook {
- boolean isEligible(QueryMeta queryMeta, LogicalPlan plan);
- void hook(QueryMeta queryMeta, LogicalPlan plan) throws Exception;
+ boolean isEligible(QueryContext queryContext, LogicalPlan plan);
+ void hook(QueryContext queryContext, LogicalPlan plan) throws Exception;
}
public class DistributedQueryHookManager {
@@ -284,11 +285,11 @@ public class GlobalEngine extends AbstractService {
hooks.add(hook);
}
- public void doHooks(QueryMeta queryMeta, LogicalPlan plan) {
+ public void doHooks(QueryContext queryContext, LogicalPlan plan) {
for (DistributedQueryHook hook : hooks) {
- if (hook.isEligible(queryMeta, plan)) {
+ if (hook.isEligible(queryContext, plan)) {
try {
- hook.hook(queryMeta, plan);
+ hook.hook(queryContext, plan);
} catch (Throwable t) {
t.printStackTrace();
}
@@ -300,7 +301,7 @@ public class GlobalEngine extends AbstractService {
private class CreateTableHook implements DistributedQueryHook {
@Override
- public boolean isEligible(QueryMeta queryMeta, LogicalPlan plan) {
+ public boolean isEligible(QueryContext queryContext, LogicalPlan plan) {
if (plan.getRootBlock().hasStoreTableNode()) {
StoreTableNode storeTableNode = plan.getRootBlock().getStoreTableNode();
return storeTableNode.isCreatedTable();
@@ -310,49 +311,49 @@ public class GlobalEngine extends AbstractService {
}
@Override
- public void hook(QueryMeta queryMeta, LogicalPlan plan) throws Exception {
+ public void hook(QueryContext queryContext, LogicalPlan plan) throws Exception {
StoreTableNode storeTableNode = plan.getRootBlock().getStoreTableNode();
- queryMeta.setOutputTable(storeTableNode.getTableName());
- queryMeta.setCreateTable();
+ String tableName = storeTableNode.getTableName();
+ queryContext.setOutputTable(tableName);
+ queryContext.setOutputPath(new Path(TajoConf.getWarehousePath(context.getConf()), tableName));
+ queryContext.setCreateTable();
}
}
private class InsertHook implements DistributedQueryHook {
@Override
- public boolean isEligible(QueryMeta queryMeta, LogicalPlan plan) {
+ public boolean isEligible(QueryContext queryContext, LogicalPlan plan) {
return plan.getRootBlock().getRootType() == NodeType.INSERT;
}
@Override
- public void hook(QueryMeta queryMeta, LogicalPlan plan) throws Exception {
- queryMeta.setInsert();
+ public void hook(QueryContext queryContext, LogicalPlan plan) throws Exception {
+ queryContext.setInsert();
InsertNode insertNode = plan.getRootBlock().getInsertNode();
StoreTableNode storeNode;
- // Set QueryMeta settings, such as output table name and output path.
+ // Set QueryContext settings, such as output table name and output path.
// It also remove data files if overwrite is true.
String outputTableName;
Path outputPath;
- if (insertNode.hasTargetTable()) {
+ if (insertNode.hasTargetTable()) { // INSERT INTO [TB_NAME]
TableDesc desc = insertNode.getTargetTable();
outputTableName = desc.getName();
outputPath = desc.getPath();
- } else {
+ queryContext.setOutputTable(outputTableName);
+ } else { // INSERT INTO LOCATION ...
outputTableName = PlannerUtil.normalizeTableName(insertNode.getPath().getName());
outputPath = insertNode.getPath();
- queryMeta.setFileOutput();
+ queryContext.setFileOutput();
}
storeNode = new StoreTableNode(outputTableName);
- queryMeta.setOutputTable(outputTableName);
- queryMeta.setOutputPath(outputPath);
+ queryContext.setOutputPath(outputPath);
- if (insertNode.isOverwrite() && sm.exists(outputPath)) {
- queryMeta.setOutputOverwrite();
- storeNode.setOverwrite();
- sm.deleteData(outputPath);
+ if (insertNode.isOverwrite()) {
+ queryContext.setOutputOverwrite();
}
////////////////////////////////////////////////////////////////////////////////////
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryContext.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryContext.java
new file mode 100644
index 0000000..8e89938
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryContext.java
@@ -0,0 +1,171 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.master;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.Options;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.engine.planner.PlannerUtil;
+import org.apache.tajo.engine.planner.logical.NodeType;
+
+import static org.apache.tajo.catalog.proto.CatalogProtos.KeyValueSetProto;
+
+public class QueryContext extends Options {
+
+ public static final String COMMAND_TYPE = "tajo.query.command";
+
+ public static final String STAGING_DIR = "tajo.query.staging_dir";
+
+ public static final String USER_NAME = "tajo.query.username";
+
+ public static final String OUTPUT_TABLE_NAME = "tajo.query.output.table";
+ public static final String OUTPUT_TABLE_PATH = "tajo.query.output.path";
+ public static final String OUTPUT_OVERWRITE = "tajo.query.output.overwrite";
+ public static final String OUTPUT_AS_DIRECTORY = "tajo.query.output.asdirectory";
+
+ public static final String TRUE_VALUE = "1";
+ public static final String FALSE_VALUE = "0";
+
+ public QueryContext() {}
+
+ public QueryContext(KeyValueSetProto proto) {
+ super(proto);
+ }
+
+ public void put(TajoConf.ConfVars key, String value) {
+ put(key.varname, value);
+ }
+
+ public String get(TajoConf.ConfVars key) {
+ return get(key.varname);
+ }
+
+ public String get(String key) {
+ return super.get(key);
+ }
+
+ public void setBool(String key, boolean val) {
+ put(key, val ? TRUE_VALUE : FALSE_VALUE);
+ }
+
+ public boolean getBool(String key) {
+ String strVal = get(key);
+ return strVal != null ? strVal.equals(TRUE_VALUE) : false;
+ }
+
+ public void setUser(String username) {
+ put(USER_NAME, username);
+ }
+
+ public String getUser() {
+ return get(USER_NAME);
+ }
+
+ public void setStagingDir(Path path) {
+ put(STAGING_DIR, path.toUri().toString());
+ }
+
+ public Path getStagingDir() {
+ String strVal = get(STAGING_DIR);
+ return strVal != null ? new Path(strVal) : null;
+ }
+
+ /**
+ * The fact that QueryContext has an output table means this query has a target table.
+ * In other words, this query is 'CREATE TABLE' or 'INSERT (OVERWRITE) INTO <table name>' statement.
+ * This config is not set if a query has INSERT (OVERWRITE) INTO LOCATION '/path/..'.
+ */
+ public boolean hasOutputTable() {
+ return get(OUTPUT_TABLE_NAME) != null;
+ }
+
+ /**
+ * Set a target table name
+ *
+ * @param tableName The target table name
+ */
+ public void setOutputTable(String tableName) {
+ put(OUTPUT_TABLE_NAME, PlannerUtil.normalizeTableName(tableName));
+ }
+
+ public String getOutputTable() {
+ String strVal = get(OUTPUT_TABLE_NAME);
+ return strVal != null ? strVal : null;
+ }
+
+ /**
+ * The fact that QueryContext has an output path means this query will write the output to a specific directory.
+ * In other words, this query is 'CREATE TABLE' or 'INSERT (OVERWRITE) INTO (<table name>|LOCATION)' statement.
+ *
+ * @return
+ */
+ public boolean hasOutputPath() {
+ return get(OUTPUT_TABLE_PATH) != null;
+ }
+
+ public void setOutputPath(Path path) {
+ put(OUTPUT_TABLE_PATH, path.toUri().toString());
+ }
+
+ public Path getOutputPath() {
+ String strVal = get(OUTPUT_TABLE_PATH);
+ return strVal != null ? new Path(strVal) : null;
+ }
+
+ public void setOutputOverwrite() {
+ setBool(OUTPUT_OVERWRITE, true);
+ }
+
+ public boolean isOutputOverwrite() {
+ return getBool(OUTPUT_OVERWRITE);
+ }
+
+ public void setFileOutput() {
+ setBool(OUTPUT_AS_DIRECTORY, true);
+ }
+
+ public boolean isFileOutput() {
+ return getBool(OUTPUT_AS_DIRECTORY);
+ }
+
+ public void setCommandType(NodeType nodeType) {
+ put(COMMAND_TYPE, nodeType.name());
+ }
+
+ public NodeType getCommandType() {
+ String strVal = get(COMMAND_TYPE);
+ return strVal != null ? NodeType.valueOf(strVal) : null;
+ }
+
+ public void setCreateTable() {
+ setCommandType(NodeType.CREATE_TABLE);
+ }
+
+ public boolean isCreateTable() {
+ return getCommandType() == NodeType.CREATE_TABLE;
+ }
+
+ public void setInsert() {
+ setCommandType(NodeType.INSERT);
+ }
+
+ public boolean isInsert() {
+ return getCommandType() == NodeType.INSERT;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java
deleted file mode 100644
index 6655317..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo.master;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.tajo.catalog.Options;
-import org.apache.tajo.conf.TajoConf;
-import org.apache.tajo.engine.planner.InsertNode;
-import org.apache.tajo.engine.planner.PlannerUtil;
-import org.apache.tajo.engine.planner.logical.CreateTableNode;
-
-import static org.apache.tajo.catalog.proto.CatalogProtos.KeyValueSetProto;
-
-public class QueryMeta extends Options {
-
- public QueryMeta() {}
-
- public QueryMeta(KeyValueSetProto proto) {
- super(proto);
- }
-
- public void put(TajoConf.ConfVars key, String value) {
- put(key.varname, value);
- }
-
- public String get(TajoConf.ConfVars key) {
- return get(key.varname);
- }
-
- public String get(String key) {
- return super.get(key);
- }
-
- public void setUser(String username) {
- put(TajoConf.ConfVars.QUERY_USERNAME, username);
- }
-
- public String getUser() {
- return get(TajoConf.ConfVars.QUERY_USERNAME);
- }
-
- public void setOutputTable(String tableName) {
- put(TajoConf.ConfVars.QUERY_OUTPUT_TABLE, PlannerUtil.normalizeTableName(tableName));
- }
-
- public String getOutputTable() {
- return get(TajoConf.ConfVars.QUERY_OUTPUT_TABLE);
- }
-
- public void setOutputPath(Path path) {
- // it is determined in QueryMaster.initStagingDir().
- put(TajoConf.ConfVars.QUERY_OUTPUT_DIR, path.toUri().toString());
- }
-
- public Path getOutputPath() {
- return new Path(get(TajoConf.ConfVars.QUERY_OUTPUT_DIR));
- }
-
- public void setOutputOverwrite() {
- put("tajo.output.overwrite", "1");
- }
-
- public boolean isOutputOverwrite() {
- String overwrite = get("tajo.output.overwrite");
- if (overwrite != null && overwrite.equals("1")) {
- return true;
- } else {
- return false;
- }
- }
-
- public void setFileOutput() {
- put("tajo.output.fileoutput", "1");
- }
-
- public boolean isFileOutput() {
- String fileoutput = get("tajo.output.fileoutput");
- if (fileoutput != null && fileoutput.equals("1")) {
- return true;
- } else {
- return false;
- }
- }
-
- public void setCreateTable() {
- put("tajo.query.command", CreateTableNode.class.getSimpleName());
- }
-
- public boolean isCreateTable() {
- String command = get("tajo.query.command");
- return command != null && command.equals(CreateTableNode.class.getSimpleName());
- }
-
- public void setInsert() {
- put("tajo.query.command", InsertNode.class.getSimpleName());
- }
-
- public boolean isInsert() {
- String command = get("tajo.query.command");
- return command != null && command.equals(InsertNode.class.getSimpleName());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoContainerProxy.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
index 966df72..5f17c71 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoContainerProxy.java
@@ -38,7 +38,7 @@ import java.util.ArrayList;
import java.util.List;
public class TajoContainerProxy extends ContainerProxy {
- public TajoContainerProxy(QueryMasterTask.QueryContext context,
+ public TajoContainerProxy(QueryMasterTask.QueryMasterTaskContext context,
Configuration conf, Container container,
ExecutionBlockId executionBlockId) {
super(context, conf, executionBlockId, container);
@@ -53,7 +53,7 @@ public class TajoContainerProxy extends ContainerProxy {
this.port = ((TajoWorkerContainer)container).getWorkerResource().getPullServerPort();
this.state = ContainerState.RUNNING;
- LOG.info("=======>Launch Container:" + executionBlockId + "," + containerID.getId() + "," +
+ LOG.info("Launch Container:" + executionBlockId + "," + containerID.getId() + "," +
container.getId() + "," + container.getNodeId() + ", pullServer=" + port);
assignExecutionBlock(executionBlockId, container);
@@ -78,7 +78,7 @@ public class TajoContainerProxy extends ContainerProxy {
.setQueryMasterPort(myAddr.getPort())
.setNodeId(container.getNodeId().toString())
.setContainerId(container.getId().toString())
- .setQueryOutputPath(context.getOutputPath().toString())
+ .setQueryOutputPath(context.getStagingDir().toString())
.build();
tajoWorkerRpcClient.executeExecutionBlock(null, request, NullCallback.get());
@@ -112,7 +112,7 @@ public class TajoContainerProxy extends ContainerProxy {
public synchronized void stopContainer() {
LOG.info("Release TajoWorker Resource: " + executionBlockId + "," + containerID + ", state:" + this.state);
if(isCompletelyDone()) {
- LOG.info("====> Container already stopped:" + containerID);
+ LOG.info("Container already stopped:" + containerID);
return;
}
if(this.state == ContainerState.PREP) {
@@ -134,7 +134,7 @@ public class TajoContainerProxy extends ContainerProxy {
}
}
- public static void releaseWorkerResource(QueryMasterTask.QueryContext context,
+ public static void releaseWorkerResource(QueryMasterTask.QueryMasterTaskContext context,
ExecutionBlockId executionBlockId,
WorkerResource workerResource) throws Exception {
List<WorkerResource> workerResources = new ArrayList<WorkerResource>();
@@ -143,7 +143,7 @@ public class TajoContainerProxy extends ContainerProxy {
releaseWorkerResource(context, executionBlockId, workerResources);
}
- public static void releaseWorkerResource(QueryMasterTask.QueryContext context,
+ public static void releaseWorkerResource(QueryMasterTask.QueryMasterTaskContext context,
ExecutionBlockId executionBlockId,
List<WorkerResource> workerResources) throws Exception {
List<TajoMasterProtocol.WorkerResourceProto> workerResourceProtos =
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
index f22472a..01f312c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -22,8 +22,10 @@ import com.google.protobuf.ServiceException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.Clock;
@@ -34,7 +36,6 @@ import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.hadoop.yarn.service.Service;
import org.apache.hadoop.yarn.util.RackResolver;
-import org.apache.tajo.TajoConstants;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos.FunctionType;
import org.apache.tajo.common.TajoDataTypes.Type;
@@ -44,11 +45,13 @@ import org.apache.tajo.engine.function.Country;
import org.apache.tajo.engine.function.InCountry;
import org.apache.tajo.engine.function.builtin.*;
import org.apache.tajo.master.querymaster.QueryJobManager;
-import org.apache.tajo.master.rm.TajoWorkerResourceManager;
import org.apache.tajo.master.rm.WorkerResourceManager;
+import org.apache.tajo.master.rm.YarnTajoResourceManager;
import org.apache.tajo.storage.StorageManager;
+import org.apache.tajo.util.NetUtils;
import org.apache.tajo.webapp.StaticHttpServer;
+import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.List;
@@ -60,12 +63,26 @@ public class TajoMaster extends CompositeService {
public static final int SHUTDOWN_HOOK_PRIORITY = 30;
+ /** rw-r--r-- */
+ final public static FsPermission TAJO_ROOT_DIR_PERMISSION = FsPermission.createImmutable((short) 0644);
+ /** rw-r--r-- */
+ final public static FsPermission SYSTEM_DIR_PERMISSION = FsPermission.createImmutable((short) 0644);
+ /** rw-r--r-- */
+ final public static FsPermission SYSTEM_RESOURCE_DIR_PERMISSION = FsPermission.createImmutable((short) 0644);
+ /** rw-r--r-- */
+ final public static FsPermission WAREHOUSE_DIR_PERMISSION = FsPermission.createImmutable((short) 0644);
+ /** rw-r--r-- */
+ final public static FsPermission STAGING_ROOTDIR_PERMISSION = FsPermission.createImmutable((short) 0644);
+ /** rw-r--r-- */
+ final public static FsPermission SYSTEM_CONF_FILE_PERMISSION = FsPermission.createImmutable((short) 0644);
+
+
private MasterContext context;
- private TajoConf conf;
+ private TajoConf systemConf;
private FileSystem defaultFS;
private Clock clock;
- private Path basePath;
+ private Path tajoRootPath;
private Path wareHousePath;
private CatalogServer catalogServer;
@@ -88,65 +105,27 @@ public class TajoMaster extends CompositeService {
@Override
public void init(Configuration _conf) {
- this.conf = (TajoConf) _conf;
+ this.systemConf = (TajoConf) _conf;
- context = new MasterContext(conf);
+ context = new MasterContext(systemConf);
clock = new SystemClock();
try {
- RackResolver.init(conf);
-
-// this.conf.writeXml(System.out);
- String className = this.conf.get("tajo.resource.manager", TajoWorkerResourceManager.class.getCanonicalName());
- Class<WorkerResourceManager> resourceManagerClass =
- (Class<WorkerResourceManager>)Class.forName(className);
-
- Constructor<WorkerResourceManager> constructor = resourceManagerClass.getConstructor(MasterContext.class);
- resourceManager = constructor.newInstance(context);
- resourceManager.init(context.getConf());
-
- //TODO WebServer port configurable
- webServer = StaticHttpServer.getInstance(this ,"admin", null, 8080 ,
- true, null, context.getConf(), null);
- webServer.start();
-
- // Get the tajo base dir
- this.basePath = new Path(conf.getVar(ConfVars.ROOT_DIR));
- LOG.info("Tajo Root dir is set " + basePath);
- // Get default DFS uri from the base dir
- this.defaultFS = basePath.getFileSystem(conf);
- conf.set("fs.defaultFS", defaultFS.getUri().toString());
- LOG.info("FileSystem (" + this.defaultFS.getUri() + ") is initialized.");
- if (!defaultFS.exists(basePath)) {
- defaultFS.mkdirs(basePath);
- LOG.info("Tajo Base dir (" + basePath + ") is created.");
- }
-
- this.storeManager = new StorageManager(conf);
-
- // Get the tajo data warehouse dir
- this.wareHousePath = new Path(basePath, TajoConstants.WAREHOUSE_DIR);
- LOG.info("Tajo Warehouse dir is set to " + wareHousePath);
- if (!defaultFS.exists(wareHousePath)) {
- defaultFS.mkdirs(wareHousePath);
- LOG.info("Warehouse dir (" + wareHousePath + ") is created");
- }
+ RackResolver.init(systemConf);
+
+ initResourceManager();
+ initWebServer();
this.dispatcher = new AsyncDispatcher();
addIfService(dispatcher);
- // The below is some mode-dependent codes
- // If tajo is local mode
- final boolean mode = conf.getBoolVar(ConfVars.CLUSTER_DISTRIBUTED);
- if (!mode) {
- LOG.info("Enabled Pseudo Distributed Mode");
- } else { // if tajo is distributed mode
- LOG.info("Enabled Distributed Mode");
- }
- // This is temporal solution of the above problem.
+ // check the system directory and create if they are not created.
+ checkAndInitializeSystemDirectories();
+ this.storeManager = new StorageManager(systemConf);
+
catalogServer = new CatalogServer(initBuiltinFunctions());
addIfService(catalogServer);
- catalog = new LocalCatalog(catalogServer);
+ catalog = new LocalCatalogWrapper(catalogServer);
globalEngine = new GlobalEngine(context);
addIfService(globalEngine);
@@ -159,12 +138,70 @@ public class TajoMaster extends CompositeService {
tajoMasterService = new TajoMasterService(context);
addIfService(tajoMasterService);
+
} catch (Exception e) {
LOG.error(e.getMessage(), e);
}
- LOG.info("====> Tajo master started");
- super.init(conf);
+ super.init(systemConf);
+
+ LOG.info("Tajo Master is initialized.");
+ }
+
+ private void initResourceManager() throws Exception {
+ Class<WorkerResourceManager> resourceManagerClass = (Class<WorkerResourceManager>)
+ systemConf.getClass(ConfVars.RESOURCE_MANAGER_CLASS.varname, YarnTajoResourceManager.class);
+ Constructor<WorkerResourceManager> constructor = resourceManagerClass.getConstructor(MasterContext.class);
+ resourceManager = constructor.newInstance(context);
+ resourceManager.init(context.getConf());
+ }
+
+ private void initWebServer() throws Exception {
+ webServer = StaticHttpServer.getInstance(this ,"admin", null, 8080 ,
+ true, null, context.getConf(), null);
+ webServer.start();
+ }
+
+ private void checkAndInitializeSystemDirectories() throws IOException {
+ // Get Tajo root dir
+ this.tajoRootPath = TajoConf.getTajoRootPath(systemConf);
+ LOG.info("Tajo Root Directory: " + tajoRootPath);
+
+ // Check and Create Tajo root dir
+ this.defaultFS = tajoRootPath.getFileSystem(systemConf);
+ systemConf.set("fs.defaultFS", defaultFS.getUri().toString());
+ LOG.info("FileSystem (" + this.defaultFS.getUri() + ") is initialized.");
+ if (!defaultFS.exists(tajoRootPath)) {
+ defaultFS.mkdirs(tajoRootPath, new FsPermission(TAJO_ROOT_DIR_PERMISSION));
+ LOG.info("Tajo Root Directory '" + tajoRootPath + "' is created.");
+ }
+
+ // Check and Create system and system resource dir
+ Path systemPath = TajoConf.getSystemPath(systemConf);
+ if (!defaultFS.exists(systemPath)) {
+ defaultFS.mkdirs(systemPath, new FsPermission(SYSTEM_DIR_PERMISSION));
+ LOG.info("System dir '" + systemPath + "' is created");
+ }
+ Path systemResourcePath = TajoConf.getSystemResourcePath(systemConf);
+ if (!defaultFS.exists(systemResourcePath)) {
+ defaultFS.mkdirs(systemResourcePath, new FsPermission(SYSTEM_RESOURCE_DIR_PERMISSION));
+ LOG.info("System resource dir '" + systemResourcePath + "' is created");
+ }
+
+ // Get Warehouse dir
+ this.wareHousePath = TajoConf.getWarehousePath(systemConf);
+ LOG.info("Tajo Warehouse Dir: " + wareHousePath);
+
+ // Check and Create Warehouse dir
+ if (!defaultFS.exists(wareHousePath)) {
+ defaultFS.mkdirs(wareHousePath, new FsPermission(WAREHOUSE_DIR_PERMISSION));
+ LOG.info("Warehouse dir '" + wareHousePath + "' is created");
+ }
+
+ Path stagingPath = TajoConf.getStagingRoot(systemConf);
+ if (!defaultFS.exists(stagingPath)) {
+ defaultFS.mkdirs(stagingPath, new FsPermission(STAGING_ROOTDIR_PERMISSION));
+ }
}
@SuppressWarnings("unchecked")
@@ -278,6 +315,33 @@ public class TajoMaster extends CompositeService {
public void start() {
LOG.info("TajoMaster startup");
super.start();
+
+ // Setting the system global configs
+ systemConf.setSocketAddr(ConfVars.CATALOG_ADDRESS.varname,
+ NetUtils.getConnectAddress(catalogServer.getBindAddress()));
+
+ try {
+ writeSystemConf();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ private void writeSystemConf() throws IOException {
+ // Storing the system configs
+ Path systemResourcePath = TajoConf.getSystemResourcePath(systemConf);
+ Path systemConfPath = new Path(systemResourcePath, "system_conf.xml");
+ systemConf.setVar(ConfVars.SYSTEM_CONF_PATH, systemConfPath.toUri().toString());
+
+ defaultFS.delete(systemConfPath, true);
+ FSDataOutputStream out = FileSystem.create(defaultFS, systemConfPath,
+ new FsPermission(SYSTEM_CONF_FILE_PERMISSION));
+ try {
+ systemConf.writeXml(out);
+ } finally {
+ out.close();
+ }
+ defaultFS.setReplication(systemConfPath, (short)systemConf.getIntVar(ConfVars.SYSTEM_CONF_REPLICA_COUNT));
}
@Override
@@ -289,17 +353,13 @@ public class TajoMaster extends CompositeService {
}
super.stop();
- LOG.info("TajoMaster main thread exiting");
+ LOG.info("Tajo Master main thread exiting");
}
public EventHandler getEventHandler() {
return dispatcher.getEventHandler();
}
- public String getMasterServerName() {
- return null;
- }
-
public boolean isMasterRunning() {
return getServiceState() == STATE.STARTED;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
index b26d05b..8bb17b1 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
@@ -52,7 +52,7 @@ public class TaskSchedulerImpl extends AbstractService
implements TaskScheduler {
private static final Log LOG = LogFactory.getLog(TaskScheduleEvent.class);
- private final QueryMasterTask.QueryContext context;
+ private final QueryMasterTask.QueryMasterTaskContext context;
private TajoAsyncDispatcher dispatcher;
private Thread eventHandlingThread;
@@ -69,7 +69,7 @@ public class TaskSchedulerImpl extends AbstractService
private int rackLocalAssigned = 0;
private int totalAssigned = 0;
- public TaskSchedulerImpl(QueryMasterTask.QueryContext context) {
+ public TaskSchedulerImpl(QueryMasterTask.QueryMasterTaskContext context) {
super(TaskSchedulerImpl.class.getName());
this.context = context;
this.dispatcher = context.getDispatcher();
@@ -228,7 +228,7 @@ public class TaskSchedulerImpl extends AbstractService
@Override
public void handle(TaskRequestEvent event) {
- LOG.info("====>TaskRequest:" + event.getContainerId() + "," + event.getExecutionBlockId());
+ LOG.info("TaskRequest: " + event.getContainerId() + "," + event.getExecutionBlockId());
if(stopEventHandling) {
event.getCallback().run(stopTaskRunnerReq);
return;
@@ -459,7 +459,7 @@ public class TaskSchedulerImpl extends AbstractService
task.getOutputName(),
false,
task.getLogicalPlan().toJson(),
- context.getQueryMeta());
+ context.getQueryContext());
if (task.getStoreTableNode().isLocal()) {
taskAssign.setInterQuery();
}
@@ -486,7 +486,7 @@ public class TaskSchedulerImpl extends AbstractService
TaskRequestEvent taskRequest;
while (it.hasNext()) {
taskRequest = it.next();
- LOG.info("====> assignToNonLeafTasks: " + taskRequest.getExecutionBlockId());
+ LOG.info("assignToNonLeafTasks: " + taskRequest.getExecutionBlockId());
QueryUnitAttemptId attemptId;
// random allocation
@@ -504,7 +504,7 @@ public class TaskSchedulerImpl extends AbstractService
task.getOutputName(),
false,
task.getLogicalPlan().toJson(),
- context.getQueryMeta());
+ context.getQueryContext());
if (task.getStoreTableNode().isLocal()) {
taskAssign.setInterQuery();
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
index 1e650e1..c4fd8b8 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
@@ -40,7 +40,6 @@ import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.ProtoUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.tajo.ExecutionBlockId;
-import org.apache.tajo.QueryConf;
import org.apache.tajo.TajoConstants;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.master.event.QueryEvent;
@@ -62,7 +61,7 @@ public class YarnContainerProxy extends ContainerProxy {
final protected String containerMgrAddress;
protected ContainerToken containerToken;
- public YarnContainerProxy(QueryMasterTask.QueryContext context, Configuration conf, YarnRPC yarnRPC,
+ public YarnContainerProxy(QueryMasterTask.QueryMasterTaskContext context, Configuration conf, YarnRPC yarnRPC,
Container container, ExecutionBlockId executionBlockId) {
super(context, conf, executionBlockId, container);
this.yarnRPC = yarnRPC;
@@ -244,9 +243,8 @@ public class YarnContainerProxy extends ContainerProxy {
if (prop.getProperty("tajo.test", "FALSE").equalsIgnoreCase("TRUE") ||
(System.getenv("tajo.test") != null && System.getenv("tajo.test").equalsIgnoreCase("TRUE"))) {
- LOG.info("=========> tajo.test is TRUE");
- environment.put(ApplicationConstants.Environment.CLASSPATH.name(), prop.getProperty(
- "java.class.path", null));
+ LOG.info("tajo.test is TRUE");
+ environment.put(ApplicationConstants.Environment.CLASSPATH.name(), prop.getProperty("java.class.path", null));
environment.put("tajo.test", "TRUE");
} else {
// Add AppMaster.jar location to classpath
@@ -297,58 +295,30 @@ public class YarnContainerProxy extends ContainerProxy {
////////////////////////////////////////////////////////////////////////////
Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
FileSystem fs = null;
-
- LOG.info("defaultFS: " + conf.get("fs.default.name"));
+ FileContext fsCtx = null;
LOG.info("defaultFS: " + conf.get("fs.defaultFS"));
try {
fs = FileSystem.get(conf);
- } catch (IOException e) {
- LOG.error(e.getMessage(), e);
- }
-
- FileContext fsCtx = null;
- try {
fsCtx = FileContext.getFileContext(conf);
- } catch (UnsupportedFileSystemException e) {
+ } catch (IOException e) {
LOG.error(e.getMessage(), e);
}
- LOG.info("Writing a QueryConf to HDFS and add to local environment");
- //Path queryConfPath = new Path(fs.getHomeDirectory(), QueryConf.FILENAME);
try {
- //writeConf(conf, queryConfPath);
- // TODO move to tajo temp
- Path warehousePath = new Path(conf.getVar(TajoConf.ConfVars.ROOT_DIR), TajoConstants.WAREHOUSE_DIR);
- Path queryConfPath = new Path(warehousePath, queryId);
- if(isMaster) {
- queryConfPath = new Path(queryConfPath, QueryConf.QUERY_MASTER_FILENAME);
- } else {
- queryConfPath = new Path(queryConfPath, QueryConf.FILENAME);
- }
-
- if(!fs.exists(queryConfPath)){
- writeConf(conf, queryConfPath);
- } else {
- LOG.warn("QueryConf already exist. path: " + queryConfPath.toString());
+ Path systemConfPath = TajoConf.getSystemConf(conf);
+ if (!fs.exists(systemConfPath)) {
+ LOG.error("system_conf.xml (" + systemConfPath.toString() + ") Not Found");
}
-
- LocalResource queryConfSrc = createApplicationResource(fsCtx,
- queryConfPath, LocalResourceType.FILE);
-// localResources.put(QueryConf.FILENAME, queryConfSrc);
- localResources.put(queryConfPath.getName(), queryConfSrc);
-
+ LocalResource systemConfResource = createApplicationResource(fsCtx, systemConfPath, LocalResourceType.FILE);
+ localResources.put(TajoConstants.SYSTEM_CONF_FILENAME, systemConfResource);
ctx.setLocalResources(localResources);
} catch (IOException e) {
LOG.error(e.getMessage(), e);
}
- // TODO - move to sub-class
- // Add shuffle token
Map<String, ByteBuffer> serviceData = new HashMap<String, ByteBuffer>();
try {
- //LOG.info("Putting shuffle token in serviceData");
- serviceData.put(PullServerAuxService.PULLSERVER_SERVICEID,
- PullServerAuxService.serializeMetaData(0));
+ serviceData.put(PullServerAuxService.PULLSERVER_SERVICEID, PullServerAuxService.serializeMetaData(0));
} catch (IOException ioe) {
LOG.error(ioe);
}
@@ -440,7 +410,7 @@ public class YarnContainerProxy extends ContainerProxy {
Vector<CharSequence> taskParams = new Vector<CharSequence>();
taskParams.add(queryMasterHost); // queryMaster hostname
taskParams.add(String.valueOf(queryMasterPort)); // queryMaster port
- taskParams.add(context.getOutputPath().toString());
+ taskParams.add(context.getStagingDir().toString());
return taskParams;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
index c4e2468..c07d759 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
@@ -47,7 +47,7 @@ public class YarnTaskRunnerLauncherImpl extends AbstractService implements TaskR
//private final YarnRPC yarnRPC;
private final static RecordFactory recordFactory =
RecordFactoryProvider.getRecordFactory(null);
- private QueryMasterTask.QueryContext context;
+ private QueryMasterTask.QueryMasterTaskContext context;
// For ContainerLauncherSpec
private static AtomicBoolean initialClasspathFlag = new AtomicBoolean();
@@ -63,7 +63,7 @@ public class YarnTaskRunnerLauncherImpl extends AbstractService implements TaskR
private YarnRPC yarnRPC;
- public YarnTaskRunnerLauncherImpl(QueryMasterTask.QueryContext context, YarnRPC yarnRPC) {
+ public YarnTaskRunnerLauncherImpl(QueryMasterTask.QueryMasterTaskContext context, YarnRPC yarnRPC) {
super(YarnTaskRunnerLauncherImpl.class.getName());
this.context = context;
this.yarnRPC = yarnRPC;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/27f76811/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
index 54cb0fa..53c08c0 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
@@ -20,7 +20,7 @@ package org.apache.tajo.master.event;
import org.apache.hadoop.yarn.event.AbstractEvent;
import org.apache.tajo.QueryId;
-import org.apache.tajo.master.QueryMeta;
+import org.apache.tajo.master.QueryContext;
/**
* This event is conveyed to QueryMaster.
@@ -31,13 +31,13 @@ public class QueryStartEvent extends AbstractEvent {
}
private QueryId queryId;
- private QueryMeta queryMeta;
+ private QueryContext queryContext;
private String logicalPlanJson;
- public QueryStartEvent(QueryId queryId, QueryMeta queryMeta, String logicalPlanJson) {
+ public QueryStartEvent(QueryId queryId, QueryContext queryContext, String logicalPlanJson) {
super(EventType.QUERY_START);
this.queryId = queryId;
- this.queryMeta = queryMeta;
+ this.queryContext = queryContext;
this.logicalPlanJson = logicalPlanJson;
}
@@ -45,8 +45,8 @@ public class QueryStartEvent extends AbstractEvent {
return queryId;
}
- public QueryMeta getQueryMeta() {
- return this.queryMeta;
+ public QueryContext getQueryContext() {
+ return this.queryContext;
}
public String getLogicalPlanJson() {