| /** |
| * 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.conf; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.tajo.ConfigKey; |
| import org.apache.tajo.SessionVars; |
| import org.apache.tajo.TajoConstants; |
| import org.apache.tajo.service.BaseServiceTracker; |
| import org.apache.tajo.util.NetUtils; |
| import org.apache.tajo.util.NumberUtil; |
| import org.apache.tajo.util.TUtil; |
| import org.apache.tajo.util.datetime.DateTimeConstants; |
| import org.apache.tajo.validation.ConstraintViolationException; |
| import org.apache.tajo.validation.Validator; |
| import org.apache.tajo.validation.Validators; |
| |
| import java.io.IOException; |
| import java.io.PrintStream; |
| import java.net.InetSocketAddress; |
| import java.util.Map; |
| import java.util.TimeZone; |
| |
| public class TajoConf extends Configuration { |
| private static TimeZone SYSTEM_TIMEZONE; |
| private static int DATE_ORDER = -1; |
| |
| private static final Map<String, ConfVars> vars = TUtil.newHashMap(); |
| |
| static { |
| Configuration.addDefaultResource("catalog-default.xml"); |
| Configuration.addDefaultResource("catalog-site.xml"); |
| Configuration.addDefaultResource("storage-default.xml"); |
| Configuration.addDefaultResource("storage-site.xml"); |
| Configuration.addDefaultResource("tajo-default.xml"); |
| Configuration.addDefaultResource("tajo-site.xml"); |
| |
| for (ConfVars confVars: ConfVars.values()) { |
| vars.put(confVars.keyname(), confVars); |
| } |
| } |
| |
| private static final String EMPTY_VALUE = ""; |
| |
| public TajoConf() { |
| super(); |
| } |
| |
| public TajoConf(Configuration conf) { |
| super(conf); |
| } |
| |
| public TajoConf(Path path) { |
| super(); |
| addResource(path); |
| } |
| |
| @SuppressWarnings("unused") |
| public TimeZone getSystemTimezone() { |
| return TimeZone.getTimeZone(getVar(ConfVars.$TIMEZONE)); |
| } |
| |
| public void setSystemTimezone(TimeZone timezone) { |
| setVar(ConfVars.$TIMEZONE, timezone.getID()); |
| } |
| |
| public static int getDateOrder() { |
| if (DATE_ORDER < 0) { |
| TajoConf tajoConf = new TajoConf(); |
| String dateOrder = tajoConf.getVar(ConfVars.$DATE_ORDER); |
| if ("YMD".equals(dateOrder)) { |
| DATE_ORDER = DateTimeConstants.DATEORDER_YMD; |
| } else if ("DMY".equals(dateOrder)) { |
| DATE_ORDER = DateTimeConstants.DATEORDER_DMY; |
| } else if ("MDY".equals(dateOrder)) { |
| DATE_ORDER = DateTimeConstants.DATEORDER_MDY; |
| } else { |
| DATE_ORDER = DateTimeConstants.DATEORDER_YMD; |
| } |
| } |
| return DATE_ORDER; |
| } |
| |
| @VisibleForTesting |
| public static int setDateOrder(int dateOrder) { |
| int oldDateOrder = DATE_ORDER; |
| DATE_ORDER = dateOrder; |
| return oldDateOrder; |
| } |
| |
| public static enum ConfVars implements ConfigKey { |
| |
| /////////////////////////////////////////////////////////////////////////////////////// |
| // Tajo System Configuration |
| // |
| // They are all static configs which are not changed or not overwritten at all. |
| /////////////////////////////////////////////////////////////////////////////////////// |
| |
| // a username for a running Tajo cluster |
| ROOT_DIR("tajo.rootdir", "file:///tmp/tajo-${user.name}/", |
| Validators.groups(Validators.notNull(), Validators.pathUrl())), |
| USERNAME("tajo.username", "${user.name}", Validators.javaString()), |
| |
| // Configurable System Directories |
| WAREHOUSE_DIR("tajo.warehouse.directory", EMPTY_VALUE, Validators.pathUrl()), |
| STAGING_ROOT_DIR("tajo.staging.directory", "/tmp/tajo-${user.name}/staging", Validators.pathUrl()), |
| |
| SYSTEM_CONF_PATH("tajo.system-conf.path", EMPTY_VALUE, Validators.pathUrl()), |
| SYSTEM_CONF_REPLICA_COUNT("tajo.system-conf.replica-count", 20, Validators.min("1")), |
| |
| // Tajo Master Service Addresses |
| TAJO_MASTER_UMBILICAL_RPC_ADDRESS("tajo.master.umbilical-rpc.address", "localhost:26001", |
| Validators.networkAddr()), |
| TAJO_MASTER_CLIENT_RPC_ADDRESS("tajo.master.client-rpc.address", "localhost:26002", |
| Validators.networkAddr()), |
| TAJO_MASTER_INFO_ADDRESS("tajo.master.info-http.address", "0.0.0.0:26080", Validators.networkAddr()), |
| |
| // High availability configurations |
| TAJO_MASTER_HA_ENABLE("tajo.master.ha.enable", false, Validators.bool()), |
| TAJO_MASTER_HA_MONITOR_INTERVAL("tajo.master.ha.monitor.interval", 5 * 1000), // 5 sec |
| TAJO_MASTER_HA_CLIENT_RETRY_MAX_NUM("tajo.master.ha.client.read.retry.max-num", 120), // 120 retry |
| TAJO_MASTER_HA_CLIENT_RETRY_PAUSE_TIME("tajo.master.ha.client.read.pause-time", 500), // 500 ms |
| |
| // Service discovery |
| DEFAULT_SERVICE_TRACKER_CLASS("tajo.discovery.service-tracker.class", BaseServiceTracker.class.getCanonicalName()), |
| HA_SERVICE_TRACKER_CLASS("tajo.discovery.ha-service-tracker.class", "org.apache.tajo.ha.HdfsServiceTracker"), |
| |
| // Resource tracker service |
| RESOURCE_TRACKER_RPC_ADDRESS("tajo.resource-tracker.rpc.address", "localhost:26003", |
| Validators.networkAddr()), |
| RESOURCE_TRACKER_HEARTBEAT_TIMEOUT("tajo.resource-tracker.heartbeat.timeout-secs", 120 * 1000), // seconds |
| |
| // QueryMaster resource |
| TAJO_QUERYMASTER_DISK_SLOT("tajo.qm.resource.disk.slots", 0.0f, Validators.min("0.0f")), |
| TAJO_QUERYMASTER_MEMORY_MB("tajo.qm.resource.memory-mb", 512, Validators.min("64")), |
| |
| // Tajo Worker Service Addresses |
| WORKER_INFO_ADDRESS("tajo.worker.info-http.address", "0.0.0.0:28080", Validators.networkAddr()), |
| WORKER_QM_INFO_ADDRESS("tajo.worker.qm-info-http.address", "0.0.0.0:28081", Validators.networkAddr()), |
| WORKER_PEER_RPC_ADDRESS("tajo.worker.peer-rpc.address", "0.0.0.0:28091", Validators.networkAddr()), |
| WORKER_CLIENT_RPC_ADDRESS("tajo.worker.client-rpc.address", "0.0.0.0:28092", Validators.networkAddr()), |
| WORKER_QM_RPC_ADDRESS("tajo.worker.qm-rpc.address", "0.0.0.0:28093", Validators.networkAddr()), |
| |
| // Tajo Worker Temporal Directories |
| WORKER_TEMPORAL_DIR("tajo.worker.tmpdir.locations", "/tmp/tajo-${user.name}/tmpdir", Validators.pathUrlList()), |
| WORKER_TEMPORAL_DIR_CLEANUP("tajo.worker.tmpdir.cleanup-at-startup", false, Validators.bool()), |
| |
| // Tajo Worker Resources |
| WORKER_RESOURCE_AVAILABLE_CPU_CORES("tajo.worker.resource.cpu-cores", 1, Validators.min("1")), |
| WORKER_RESOURCE_AVAILABLE_MEMORY_MB("tajo.worker.resource.memory-mb", 1024, Validators.min("64")), |
| WORKER_RESOURCE_AVAILABLE_DISKS("tajo.worker.resource.disks", 1.0f), |
| WORKER_EXECUTION_MAX_SLOTS("tajo.worker.parallel-execution.max-num", 2), |
| WORKER_RESOURCE_DFS_DIR_AWARE("tajo.worker.resource.dfs-dir-aware", false, Validators.bool()), |
| |
| // Tajo Worker Dedicated Resources |
| WORKER_RESOURCE_DEDICATED("tajo.worker.resource.dedicated", false, Validators.bool()), |
| WORKER_RESOURCE_DEDICATED_MEMORY_RATIO("tajo.worker.resource.dedicated-memory-ratio", 0.8f, |
| Validators.range("0.0f", "1.0f")), |
| |
| // Tajo History |
| WORKER_HISTORY_EXPIRE_PERIOD("tajo.worker.history.expire-interval-minutes", 60), // 1 hours |
| QUERYMASTER_HISTORY_EXPIRE_PERIOD("tajo.qm.history.expire-interval-minutes", 6 * 60), // 6 hours |
| |
| WORKER_HEARTBEAT_TIMEOUT("tajo.worker.heartbeat.timeout", 120 * 1000), // 120 sec |
| |
| // Resource Manager |
| RESOURCE_MANAGER_CLASS("tajo.resource.manager", "org.apache.tajo.master.rm.TajoWorkerResourceManager", |
| Validators.groups(Validators.notNull(), Validators.clazz())), |
| |
| // Catalog |
| CATALOG_ADDRESS("tajo.catalog.client-rpc.address", "localhost:26005", Validators.networkAddr()), |
| |
| |
| // for Yarn Resource Manager ---------------------------------------------- |
| |
| /** how many launching TaskRunners in parallel */ |
| YARN_RM_QUERY_MASTER_MEMORY_MB("tajo.querymaster.memory-mb", 512, Validators.min("64")), |
| YARN_RM_QUERY_MASTER_DISKS("tajo.yarn-rm.querymaster.disks", 1), |
| YARN_RM_TASKRUNNER_LAUNCH_PARALLEL_NUM("tajo.yarn-rm.parallel-task-runner-launcher-num", |
| Runtime.getRuntime().availableProcessors() * 2), |
| YARN_RM_WORKER_NUMBER_PER_NODE("tajo.yarn-rm.max-worker-num-per-node", 8), |
| |
| // Query Configuration |
| QUERY_SESSION_TIMEOUT("tajo.query.session.timeout-sec", 60, Validators.min("0")), |
| |
| // Shuffle Configuration -------------------------------------------------- |
| PULLSERVER_PORT("tajo.pullserver.port", 0, Validators.range("0", "65535")), |
| SHUFFLE_SSL_ENABLED_KEY("tajo.pullserver.ssl.enabled", false, Validators.bool()), |
| SHUFFLE_FILE_FORMAT("tajo.shuffle.file-format", "RAW"), |
| SHUFFLE_FETCHER_PARALLEL_EXECUTION_MAX_NUM("tajo.shuffle.fetcher.parallel-execution.max-num", 2), |
| SHUFFLE_FETCHER_CHUNK_MAX_SIZE("tajo.shuffle.fetcher.chunk.max-size", 8192), |
| SHUFFLE_FETCHER_READ_TIMEOUT("tajo.shuffle.fetcher.read.timeout-sec", 120), |
| SHUFFLE_FETCHER_READ_RETRY_MAX_NUM("tajo.shuffle.fetcher.read.retry.max-num", 20), |
| SHUFFLE_HASH_APPENDER_BUFFER_SIZE("tajo.shuffle.hash.appender.buffer.size", 10000), |
| SHUFFLE_HASH_APPENDER_PAGE_VOLUME("tajo.shuffle.hash.appender.page.volumn-mb", 30), |
| HASH_SHUFFLE_PARENT_DIRS("tajo.hash.shuffle.parent.dirs.count", 10), |
| |
| // Storage Configuration -------------------------------------------------- |
| ROWFILE_SYNC_INTERVAL("rowfile.sync.interval", 100), |
| MINIMUM_SPLIT_SIZE("tajo.min.split.size", (long) 1), |
| // for RCFile |
| HIVEUSEEXPLICITRCFILEHEADER("tajo.exec.rcfile.use.explicit.header", true, Validators.bool()), |
| |
| // RPC -------------------------------------------------------------------- |
| RPC_POOL_MAX_IDLE("tajo.rpc.pool.idle.max", 10), |
| |
| // Internal RPC Client |
| INTERNAL_RPC_CLIENT_WORKER_THREAD_NUM("tajo.internal.rpc.client.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 2), |
| |
| // Internal RPC Server |
| MASTER_RPC_SERVER_WORKER_THREAD_NUM("tajo.master.rpc.server.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 2), |
| QUERY_MASTER_RPC_SERVER_WORKER_THREAD_NUM("tajo.querymaster.rpc.server.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 2), |
| WORKER_RPC_SERVER_WORKER_THREAD_NUM("tajo.worker.rpc.server.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 2), |
| CATALOG_RPC_SERVER_WORKER_THREAD_NUM("tajo.catalog.rpc.server.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 2), |
| SHUFFLE_RPC_SERVER_WORKER_THREAD_NUM("tajo.shuffle.rpc.server.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 2), |
| |
| // Client RPC |
| RPC_CLIENT_WORKER_THREAD_NUM("tajo.rpc.client.worker-thread-num", 4), |
| |
| SHUFFLE_RPC_CLIENT_WORKER_THREAD_NUM("tajo.shuffle.rpc.client.worker-thread-num", |
| Runtime.getRuntime().availableProcessors()), |
| |
| //Client service RPC Server |
| MASTER_SERVICE_RPC_SERVER_WORKER_THREAD_NUM("tajo.master.service.rpc.server.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 1), |
| WORKER_SERVICE_RPC_SERVER_WORKER_THREAD_NUM("tajo.worker.service.rpc.server.worker-thread-num", |
| Runtime.getRuntime().availableProcessors() * 1), |
| |
| // Task Configuration ----------------------------------------------------- |
| TASK_DEFAULT_MEMORY("tajo.task.memory-slot-mb.default", 512), |
| TASK_DEFAULT_DISK("tajo.task.disk-slot.default", 0.5f), |
| TASK_DEFAULT_SIZE("tajo.task.size-mb", 128), |
| |
| // Query and Optimization ------------------------------------------------- |
| // This class provides a ordered list of logical plan rewrite rule classes. |
| LOGICAL_PLAN_REWRITE_RULE_PROVIDER_CLASS("tajo.plan.logical.rewriter.provider", |
| "org.apache.tajo.plan.rewrite.BaseLogicalPlanRewriteRuleProvider"), |
| // This class provides a ordered list of global plan rewrite rule classes. |
| GLOBAL_PLAN_REWRITE_RULE_PROVIDER_CLASS("tajo.plan.global.rewriter.provider", |
| "org.apache.tajo.engine.planner.global.rewriter.BaseGlobalPlanRewriteRuleProvider"), |
| EXECUTOR_EXTERNAL_SORT_THREAD_NUM("tajo.executor.external-sort.thread-num", 1), |
| EXECUTOR_EXTERNAL_SORT_FANOUT("tajo.executor.external-sort.fanout-num", 8), |
| |
| EXECUTOR_INNER_JOIN_INMEMORY_HASH_TABLE_SIZE("tajo.executor.join.inner.in-memory-table-num", (long)1000000), |
| |
| // Metrics ---------------------------------------------------------------- |
| METRICS_PROPERTY_FILENAME("tajo.metrics.property.file", "tajo-metrics.properties"), |
| |
| // Query History --------------------------------------------------------- |
| HISTORY_QUERY_DIR("tajo.history.query.dir", STAGING_ROOT_DIR.defaultVal + "/history"), |
| HISTORY_TASK_DIR("tajo.history.task.dir", "file:///tmp/tajo-${user.name}/history"), |
| HISTORY_EXPIRY_TIME_DAY("tajo.history.expiry-time-day", 7), |
| HISTORY_QUERY_REPLICATION("tajo.history.query.replication", 1, Validators.min("1")), |
| HISTORY_TASK_REPLICATION("tajo.history.task.replication", 1, Validators.min("1")), |
| |
| // Misc ------------------------------------------------------------------- |
| // Fragment |
| // When making physical plan, the length of fragment is used to determine the physical operation. |
| // Some storage does not know the size of the fragment. |
| // In this case PhysicalPlanner uses this value to determine. |
| FRAGMENT_ALTERNATIVE_UNKNOWN_LENGTH("tajo.fragment.alternative.unknown.length", (long)(512 * 1024 * 1024)), |
| |
| // Geo IP |
| GEOIP_DATA("tajo.function.geoip-database-location", ""), |
| |
| ///////////////////////////////////////////////////////////////////////////////// |
| // User Session Configuration |
| // |
| // All session variables begin with dollor($) sign. They are default configs |
| // for session variables. Do not directly use the following configs. Instead, |
| // please use QueryContext in order to access session variables. |
| // |
| // Also, users can change the default values of session variables in tajo-site.xml. |
| ///////////////////////////////////////////////////////////////////////////////// |
| |
| |
| $EMPTY("tajo._", ""), |
| |
| // Query and Optimization --------------------------------------------------- |
| |
| // for distributed query strategies |
| $DIST_QUERY_BROADCAST_JOIN_THRESHOLD("tajo.dist-query.join.broadcast.threshold-bytes", (long)5 * 1048576), |
| |
| $DIST_QUERY_JOIN_TASK_VOLUME("tajo.dist-query.join.task-volume-mb", 128), |
| $DIST_QUERY_SORT_TASK_VOLUME("tajo.dist-query.sort.task-volume-mb", 128), |
| $DIST_QUERY_GROUPBY_TASK_VOLUME("tajo.dist-query.groupby.task-volume-mb", 128), |
| |
| $DIST_QUERY_JOIN_PARTITION_VOLUME("tajo.dist-query.join.partition-volume-mb", 128, Validators.min("1")), |
| $DIST_QUERY_GROUPBY_PARTITION_VOLUME("tajo.dist-query.groupby.partition-volume-mb", 256, Validators.min("1")), |
| $DIST_QUERY_TABLE_PARTITION_VOLUME("tajo.dist-query.table-partition.task-volume-mb", 256, Validators.min("1")), |
| |
| $GROUPBY_MULTI_LEVEL_ENABLED("tajo.dist-query.groupby.multi-level-aggr", true), |
| |
| // for physical Executors |
| $EXECUTOR_EXTERNAL_SORT_BUFFER_SIZE("tajo.executor.external-sort.buffer-mb", 200L), |
| $EXECUTOR_HASH_JOIN_SIZE_THRESHOLD("tajo.executor.join.common.in-memory-hash-threshold-bytes", |
| (long)256 * 1048576), |
| $EXECUTOR_INNER_HASH_JOIN_SIZE_THRESHOLD("tajo.executor.join.inner.in-memory-hash-threshold-bytes", |
| (long)256 * 1048576), |
| $EXECUTOR_OUTER_HASH_JOIN_SIZE_THRESHOLD("tajo.executor.join.outer.in-memory-hash-threshold-bytes", |
| (long)256 * 1048576), |
| $EXECUTOR_GROUPBY_INMEMORY_HASH_THRESHOLD("tajo.executor.groupby.in-memory-hash-threshold-bytes", |
| (long)256 * 1048576), |
| $MAX_OUTPUT_FILE_SIZE("tajo.query.max-outfile-size-mb", 0), // zero means infinite |
| $CODEGEN("tajo.executor.codegen.enabled", false), // Runtime code generation |
| |
| // Client ----------------------------------------------------------------- |
| $CLIENT_SESSION_EXPIRY_TIME("tajo.client.session.expiry-time-sec", 3600), // default time is one hour. |
| |
| // Command line interface and its behavior -------------------------------- |
| $CLI_MAX_COLUMN("tajo.cli.max_columns", 120), |
| $CLI_NULL_CHAR("tajo.cli.nullchar", ""), |
| $CLI_PRINT_PAUSE_NUM_RECORDS("tajo.cli.print.pause.num.records", 100), |
| $CLI_PRINT_PAUSE("tajo.cli.print.pause", true), |
| $CLI_PRINT_ERROR_TRACE("tajo.cli.print.error.trace", true), |
| $CLI_OUTPUT_FORMATTER_CLASS("tajo.cli.output.formatter", "org.apache.tajo.cli.tsql.DefaultTajoCliOutputFormatter"), |
| $CLI_ERROR_STOP("tajo.cli.error.stop", false), |
| |
| // Timezone & Date ---------------------------------------------------------- |
| $TIMEZONE("tajo.timezone", TimeZone.getDefault().getID()), |
| $DATE_ORDER("tajo.datetime.date-order", "YMD"), |
| |
| // FILE FORMAT |
| $TEXT_NULL("tajo.text.null", "\\\\N"), |
| |
| // Only for Debug and Testing |
| $DEBUG_ENABLED("tajo.debug.enabled", false), |
| $TEST_BROADCAST_JOIN_ENABLED("tajo.dist-query.join.auto-broadcast", true), |
| $TEST_JOIN_OPT_ENABLED("tajo.test.plan.join-optimization.enabled", true), |
| $TEST_FILTER_PUSHDOWN_ENABLED("tajo.test.plan.filter-pushdown.enabled", true), |
| $TEST_MIN_TASK_NUM("tajo.test.min-task-num", -1), |
| |
| // Behavior Control --------------------------------------------------------- |
| $BEHAVIOR_ARITHMETIC_ABORT("tajo.behavior.arithmetic-abort", false), |
| |
| // If True, a partitioned table is overwritten even if a sub query leads to no result. |
| // Otherwise, the table data will be kept if there is no result |
| $PARTITION_NO_RESULT_OVERWRITE_ENABLED("tajo.partition.overwrite.even-if-no-result", false), |
| |
| // ResultSet --------------------------------------------------------- |
| $RESULT_SET_FETCH_ROWNUM("tajo.resultset.fetch.rownum", 200), |
| $RESULT_SET_BLOCK_WAIT("tajo.resultset.block.wait", true), |
| ; |
| |
| public final String varname; |
| public final String defaultVal; |
| public final int defaultIntVal; |
| public final long defaultLongVal; |
| public final float defaultFloatVal; |
| public final Class<?> valClass; |
| public final boolean defaultBoolVal; |
| |
| private final VarType type; |
| private Validator validator; |
| |
| ConfVars(String varname, String defaultVal) { |
| this.varname = varname; |
| this.valClass = String.class; |
| this.defaultVal = defaultVal; |
| this.defaultIntVal = -1; |
| this.defaultLongVal = -1; |
| this.defaultFloatVal = -1; |
| this.defaultBoolVal = false; |
| this.type = VarType.STRING; |
| } |
| |
| ConfVars(String varname, String defaultVal, Validator validator) { |
| this(varname, defaultVal); |
| this.validator = validator; |
| } |
| |
| ConfVars(String varname, int defaultIntVal) { |
| this.varname = varname; |
| this.valClass = Integer.class; |
| this.defaultVal = Integer.toString(defaultIntVal); |
| this.defaultIntVal = defaultIntVal; |
| this.defaultLongVal = -1; |
| this.defaultFloatVal = -1; |
| this.defaultBoolVal = false; |
| this.type = VarType.INT; |
| } |
| |
| ConfVars(String varname, int defaultIntVal, Validator validator) { |
| this(varname, defaultIntVal); |
| this.validator = validator; |
| } |
| |
| ConfVars(String varname, long defaultLongVal) { |
| this.varname = varname; |
| this.valClass = Long.class; |
| this.defaultVal = Long.toString(defaultLongVal); |
| this.defaultIntVal = -1; |
| this.defaultLongVal = defaultLongVal; |
| this.defaultFloatVal = -1; |
| this.defaultBoolVal = false; |
| this.type = VarType.LONG; |
| } |
| |
| ConfVars(String varname, long defaultLongVal, Validator validator) { |
| this(varname, defaultLongVal); |
| this.validator = validator; |
| } |
| |
| ConfVars(String varname, float defaultFloatVal) { |
| this.varname = varname; |
| this.valClass = Float.class; |
| this.defaultVal = Float.toString(defaultFloatVal); |
| this.defaultIntVal = -1; |
| this.defaultLongVal = -1; |
| this.defaultFloatVal = defaultFloatVal; |
| this.defaultBoolVal = false; |
| this.type = VarType.FLOAT; |
| } |
| |
| ConfVars(String varname, float defaultFloatVal, Validator validator) { |
| this(varname, defaultFloatVal); |
| this.validator = validator; |
| } |
| |
| ConfVars(String varname, boolean defaultBoolVal) { |
| this.varname = varname; |
| this.valClass = Boolean.class; |
| this.defaultVal = Boolean.toString(defaultBoolVal); |
| this.defaultIntVal = -1; |
| this.defaultLongVal = -1; |
| this.defaultFloatVal = -1; |
| this.defaultBoolVal = defaultBoolVal; |
| this.type = VarType.BOOLEAN; |
| } |
| |
| ConfVars(String varname, boolean defaultBoolVal, Validator validator) { |
| this(varname, defaultBoolVal); |
| this.validator = validator; |
| } |
| |
| enum VarType { |
| STRING { void checkType(String value) throws Exception { } }, |
| INT { void checkType(String value) throws Exception { Integer.valueOf(value); } }, |
| LONG { void checkType(String value) throws Exception { Long.valueOf(value); } }, |
| FLOAT { void checkType(String value) throws Exception { Float.valueOf(value); } }, |
| BOOLEAN { void checkType(String value) throws Exception { Boolean.valueOf(value); } }; |
| |
| boolean isType(String value) { |
| try { checkType(value); } catch (Exception e) { return false; } |
| return true; |
| } |
| String typeString() { return name().toUpperCase();} |
| abstract void checkType(String value) throws Exception; |
| } |
| |
| @Override |
| public String keyname() { |
| return varname; |
| } |
| |
| @Override |
| public ConfigType type() { |
| return ConfigType.SYSTEM; |
| } |
| |
| @Override |
| public Class<?> valueClass() { |
| return valClass; |
| } |
| |
| @Override |
| public Validator validator() { |
| return validator; |
| } |
| } |
| |
| public static int getIntVar(Configuration conf, ConfVars var) { |
| assert (var.valClass == Integer.class); |
| return conf.getInt(var.varname, var.defaultIntVal); |
| } |
| |
| public static void setIntVar(Configuration conf, ConfVars var, int val) { |
| assert (var.valClass == Integer.class); |
| conf.setInt(var.varname, val); |
| } |
| |
| public int getIntVar(ConfVars var) { |
| return getIntVar(this, var); |
| } |
| |
| public void setIntVar(ConfVars var, int val) { |
| setIntVar(this, var, val); |
| } |
| |
| public static long getLongVar(Configuration conf, ConfVars var) { |
| assert (var.valClass == Long.class || var.valClass == Integer.class); |
| if (var.valClass == Integer.class) { |
| return conf.getInt(var.varname, var.defaultIntVal); |
| } else { |
| return conf.getLong(var.varname, var.defaultLongVal); |
| } |
| } |
| |
| public static long getLongVar(Configuration conf, ConfVars var, long defaultVal) { |
| return conf.getLong(var.varname, defaultVal); |
| } |
| |
| public static void setLongVar(Configuration conf, ConfVars var, long val) { |
| assert (var.valClass == Long.class); |
| conf.setLong(var.varname, val); |
| } |
| |
| public long getLongVar(ConfVars var) { |
| return getLongVar(this, var); |
| } |
| |
| public void setLongVar(ConfVars var, long val) { |
| setLongVar(this, var, val); |
| } |
| |
| public static float getFloatVar(Configuration conf, ConfVars var) { |
| assert (var.valClass == Float.class); |
| return conf.getFloat(var.varname, var.defaultFloatVal); |
| } |
| |
| public static float getFloatVar(Configuration conf, ConfVars var, float defaultVal) { |
| return conf.getFloat(var.varname, defaultVal); |
| } |
| |
| public static void setFloatVar(Configuration conf, ConfVars var, float val) { |
| assert (var.valClass == Float.class); |
| conf.setFloat(var.varname, val); |
| } |
| |
| public float getFloatVar(ConfVars var) { |
| return getFloatVar(this, var); |
| } |
| |
| public void setFloatVar(ConfVars var, float val) { |
| setFloatVar(this, var, val); |
| } |
| |
| public static boolean getBoolVar(Configuration conf, ConfVars var) { |
| assert (var.valClass == Boolean.class); |
| return conf.getBoolean(var.varname, var.defaultBoolVal); |
| } |
| |
| public static boolean getBoolVar(Configuration conf, ConfVars var, boolean defaultVal) { |
| return conf.getBoolean(var.varname, defaultVal); |
| } |
| |
| public static void setBoolVar(Configuration conf, ConfVars var, boolean val) { |
| assert (var.valClass == Boolean.class); |
| conf.setBoolean(var.varname, val); |
| } |
| |
| public boolean getBoolVar(ConfVars var) { |
| return getBoolVar(this, var); |
| } |
| |
| public void setBoolVar(ConfVars var, boolean val) { |
| setBoolVar(this, var, val); |
| } |
| |
| public void setClassVar(ConfVars var, Class<?> clazz) { |
| setVar(var, clazz.getCanonicalName()); |
| } |
| |
| public Class<?> getClassVar(ConfVars var) { |
| String valueString = getVar(var); |
| |
| try { |
| return getClassByName(valueString); |
| } catch (ClassNotFoundException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| public static String getVar(Configuration conf, ConfVars var) { |
| return conf.get(var.varname, var.defaultVal); |
| } |
| |
| public static String getVar(Configuration conf, ConfVars var, String defaultVal) { |
| return conf.get(var.varname, defaultVal); |
| } |
| |
| public static void setVar(Configuration conf, ConfVars var, String val) { |
| assert (var.valClass == String.class); |
| conf.set(var.varname, val); |
| } |
| |
| public static ConfVars getConfVars(String name) { |
| return vars.get(name); |
| } |
| |
| public String getVar(ConfVars var) { |
| return getVar(this, var); |
| } |
| |
| public void setVar(ConfVars var, String val) { |
| setVar(this, var, val); |
| } |
| |
| public void logVars(PrintStream ps) { |
| for (ConfVars one : ConfVars.values()) { |
| ps.println(one.varname + "=" + ((get(one.varname) != null) ? get(one.varname) : "")); |
| } |
| } |
| |
| public InetSocketAddress getSocketAddrVar(ConfVars var) { |
| final String address = getVar(var); |
| return NetUtils.createSocketAddr(address); |
| } |
| |
| ///////////////////////////////////////////////////////////////////////////// |
| // Tajo System Specific Methods |
| ///////////////////////////////////////////////////////////////////////////// |
| |
| public static Path getTajoRootDir(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 getWarehouseDir(TajoConf conf) { |
| String warehousePath = conf.getVar(ConfVars.WAREHOUSE_DIR); |
| if (warehousePath == null || warehousePath.equals("")) { |
| Path rootDir = getTajoRootDir(conf); |
| warehousePath = new Path(rootDir, TajoConstants.WAREHOUSE_DIR_NAME).toUri().toString(); |
| conf.setVar(ConfVars.WAREHOUSE_DIR, warehousePath); |
| return new Path(warehousePath); |
| } else { |
| return new Path(warehousePath); |
| } |
| } |
| |
| public static Path getSystemDir(TajoConf conf) { |
| Path rootPath = getTajoRootDir(conf); |
| return new Path(rootPath, TajoConstants.SYSTEM_DIR_NAME); |
| } |
| |
| public static Path getSystemResourceDir(TajoConf conf) { |
| return new Path(getSystemDir(conf), TajoConstants.SYSTEM_RESOURCE_DIR_NAME); |
| } |
| |
| public static Path getSystemHADir(TajoConf conf) { |
| return new Path(getSystemDir(conf), TajoConstants.SYSTEM_HA_DIR_NAME); |
| } |
| |
| private static boolean hasScheme(String path) { |
| return path.indexOf("file:/") == 0 || path.indexOf("hdfs:/") == 0; |
| } |
| |
| /** |
| * It returns the default root staging directory used by queries without a target table or |
| * a specified output directory. An example query is <pre>SELECT a,b,c FROM XXX;</pre>. |
| * |
| * @param conf TajoConf |
| * @return Path which points the default staging directory |
| * @throws IOException |
| */ |
| public static Path getDefaultRootStagingDir(TajoConf conf) throws IOException { |
| String stagingDirString = conf.getVar(ConfVars.STAGING_ROOT_DIR); |
| if (!hasScheme(stagingDirString)) { |
| Path warehousePath = getWarehouseDir(conf); |
| FileSystem fs = warehousePath.getFileSystem(conf); |
| Path path = new Path(fs.getUri().toString(), stagingDirString); |
| conf.setVar(ConfVars.STAGING_ROOT_DIR, path.toString()); |
| return path; |
| } |
| return new Path(stagingDirString); |
| } |
| |
| public static Path getQueryHistoryDir(TajoConf conf) throws IOException { |
| String historyDirString = conf.getVar(ConfVars.HISTORY_QUERY_DIR); |
| if (!hasScheme(historyDirString)) { |
| Path stagingPath = getDefaultRootStagingDir(conf); |
| FileSystem fs = stagingPath.getFileSystem(conf); |
| Path path = new Path(fs.getUri().toString(), historyDirString); |
| conf.setVar(ConfVars.HISTORY_QUERY_DIR, path.toString()); |
| return path; |
| } |
| return new Path(historyDirString); |
| } |
| |
| public static Path getTaskHistoryDir(TajoConf conf) throws IOException { |
| String historyDirString = conf.getVar(ConfVars.HISTORY_TASK_DIR); |
| if (!hasScheme(historyDirString)) { |
| //Local dir |
| historyDirString = "file://" + historyDirString; |
| } |
| return new Path(historyDirString); |
| } |
| |
| public static Path getSystemConfPath(TajoConf conf) { |
| String systemConfPathStr = conf.getVar(ConfVars.SYSTEM_CONF_PATH); |
| if (systemConfPathStr == null || systemConfPathStr.equals("")) { |
| Path systemResourcePath = getSystemResourceDir(conf); |
| Path systemConfPath = new Path(systemResourcePath, TajoConstants.SYSTEM_CONF_FILENAME); |
| conf.setVar(ConfVars.SYSTEM_CONF_PATH, systemConfPath.toString()); |
| return systemConfPath; |
| } else { |
| return new Path(systemConfPathStr); |
| } |
| } |
| |
| /** |
| * validateProperty function will fetch pre-defined configuration property by keyname. |
| * If found, it will validate the supplied value with these validators. |
| * |
| * @param name - a string containing specific key |
| * @param value - a string containing value |
| * @throws ConstraintViolationException |
| */ |
| public void validateProperty(String name, String value) throws ConstraintViolationException { |
| ConfigKey configKey = null; |
| configKey = TajoConf.getConfVars(name); |
| if (configKey == null) { |
| configKey = SessionVars.get(name); |
| } |
| if (configKey != null && configKey.validator() != null && configKey.valueClass() != null) { |
| Object valueObj = value; |
| if (Number.class.isAssignableFrom(configKey.valueClass())) { |
| valueObj = NumberUtil.numberValue(configKey.valueClass(), value); |
| if (valueObj == null) { |
| return; |
| } |
| } |
| configKey.validator().validate(valueObj, true); |
| } |
| } |
| |
| } |
