diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Compiler.java b/ql/src/java/org/apache/hadoop/hive/ql/Compiler.java index a559d9064c..09c838edca 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/Compiler.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/Compiler.java @@ -418,8 +418,13 @@ private void authorize(BaseSemanticAnalyzer sem) throws HiveException, CommandPr private void explainOutput(BaseSemanticAnalyzer sem, QueryPlan plan) throws IOException { if (driverContext.getConf().getBoolVar(ConfVars.HIVE_LOG_EXPLAIN_OUTPUT) || driverContext.getConf().getBoolVar(ConfVars.HIVE_SERVER2_WEBUI_EXPLAIN_OUTPUT)) { - String explainOutput = ExplainTask.getExplainOutput(sem, plan, tree, driverContext.getQueryState(), - context, driverContext.getConf()); + String explainOutput = null; + try { + explainOutput = ExplainTask.getExplainOutput(sem, plan, tree, driverContext.getQueryState(), + context, driverContext.getConf()); + } catch (Exception e) { + LOG.warn("Failed to generate explain output", e); + } if (explainOutput != null) { if (driverContext.getConf().getBoolVar(ConfVars.HIVE_LOG_EXPLAIN_OUTPUT)) { LOG.info("EXPLAIN output for queryid " + driverContext.getQueryId() + " : " + explainOutput); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsUpdateTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsUpdateTask.java index dc6d31a9cb..ab69a5fe6b 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsUpdateTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsUpdateTask.java @@ -57,8 +57,6 @@ import org.apache.hadoop.hive.ql.plan.api.StageType; import org.apache.hadoop.hive.serde2.io.DateWritableV2; import org.apache.hadoop.hive.serde2.io.TimestampWritableV2; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * ColumnStatsUpdateTask implementation. For example, ALTER TABLE src_stat @@ -70,8 +68,6 @@ public class ColumnStatsUpdateTask extends Task { private static final long serialVersionUID = 1L; - private static transient final Logger LOG = LoggerFactory - .getLogger(ColumnStatsUpdateTask.class); private ColumnStatistics constructColumnStatsFromInput() throws SemanticException, MetaException { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainSQRewriteTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainSQRewriteTask.java index 1929fab943..99a8daf939 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainSQRewriteTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainSQRewriteTask.java @@ -37,12 +37,9 @@ import org.apache.hadoop.hive.ql.plan.ExplainSQRewriteWork; import org.apache.hadoop.hive.ql.plan.api.StageType; import org.apache.hadoop.io.IOUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; public class ExplainSQRewriteTask extends Task implements Serializable { private static final long serialVersionUID = 1L; - private final Logger LOG = LoggerFactory.getLogger(this.getClass().getName()); @Override public StageType getType() { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java index c1f94d165b..6dc3067ab3 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java @@ -77,8 +77,6 @@ import org.json.JSONArray; import org.json.JSONException; import org.json.JSONObject; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import com.google.common.annotations.VisibleForTesting; @@ -87,8 +85,6 @@ * **/ public class ExplainTask extends Task implements Serializable { - private static final Logger LOG = LoggerFactory.getLogger(ExplainTask.class.getName()); - public static final String STAGE_DEPENDENCIES = "STAGE DEPENDENCIES"; private static final long serialVersionUID = 1L; public static final String EXPL_COLUMN_NAME = "Explain"; @@ -1304,7 +1300,7 @@ public boolean canExecuteInParallel() { * @param astTree AST tree dump */ public static String getExplainOutput(BaseSemanticAnalyzer sem, QueryPlan plan, ASTNode astTree, - QueryState queryState, Context context, HiveConf conf) throws IOException { + QueryState queryState, Context context, HiveConf conf) throws Exception { String ret = null; ExplainTask task = new ExplainTask(); task.initialize(queryState, plan, null, context); @@ -1329,7 +1325,7 @@ public static String getExplainOutput(BaseSemanticAnalyzer sem, QueryPlan plan, ret = baos.toString(); } } catch (Exception e) { - LOG.warn("Exception generating explain output: " + e, e); + throw e; } return ret; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ExportTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ExportTask.java index 56f0c9308e..780b6f10de 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ExportTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ExportTask.java @@ -21,14 +21,11 @@ Licensed to the Apache Software Foundation (ASF) under one import org.apache.hadoop.hive.ql.parse.repl.dump.TableExport; import org.apache.hadoop.hive.ql.plan.ExportWork; import org.apache.hadoop.hive.ql.plan.api.StageType; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.Serializable; public class ExportTask extends Task implements Serializable { private static final long serialVersionUID = 1L; - private Logger LOG = LoggerFactory.getLogger(ExportTask.class); public ExportTask() { super(); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java index 502d41d44d..95d78bf03a 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java @@ -22,8 +22,6 @@ import java.io.Serializable; import java.util.List; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.apache.hadoop.hive.ql.Context; import org.apache.hadoop.hive.ql.TaskQueue; import org.apache.hadoop.hive.ql.QueryPlan; @@ -50,7 +48,6 @@ private FetchOperator fetch; private ListSinkOperator sink; private int totalRows; - private static transient final Logger LOG = LoggerFactory.getLogger(FetchTask.class); JobConf job = null; public FetchTask() { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java index 51de87f2fd..ed40dab508 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java @@ -63,8 +63,6 @@ import org.apache.hadoop.hive.ql.session.SessionState; import org.apache.hadoop.hive.ql.util.DirectionUtils; import org.apache.hadoop.util.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.Serializable; @@ -81,7 +79,6 @@ public class MoveTask extends Task implements Serializable { private static final long serialVersionUID = 1L; - private static transient final Logger LOG = LoggerFactory.getLogger(MoveTask.class); public MoveTask() { super(); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ReplCopyTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ReplCopyTask.java index 3c7274cb4f..bbe9a8a66a 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ReplCopyTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ReplCopyTask.java @@ -56,7 +56,7 @@ private static final long serialVersionUID = 1L; - private static transient final Logger LOG = LoggerFactory.getLogger(ReplCopyTask.class); + private static transient final Logger LOAD_LOG = LoggerFactory.getLogger(ReplCopyTask.class); public ReplCopyTask(){ super(); @@ -332,7 +332,7 @@ public String getName() { HiveConf conf, boolean isAutoPurge, boolean needRecycle, boolean copyToMigratedTxnTable, boolean readSourceAsFileList) { Task copyTask = null; - LOG.debug("ReplCopyTask:getLoadCopyTask: {}=>{}", srcPath, dstPath); + LOAD_LOG.debug("ReplCopyTask:getLoadCopyTask: {}=>{}", srcPath, dstPath); if ((replicationSpec != null) && replicationSpec.isInReplicationScope()){ ReplCopyWork rcwork = new ReplCopyWork(srcPath, dstPath, false); rcwork.setReadSrcAsFilesList(readSourceAsFileList); @@ -346,12 +346,12 @@ public String getName() { // data invisible. Doing duplicate check and ignoring copy will cause consistency issue if there are multiple // replace events getting replayed in the first incremental load. rcwork.setCheckDuplicateCopy(replicationSpec.needDupCopyCheck() && !replicationSpec.isReplace()); - LOG.debug("ReplCopyTask:\trcwork"); + LOAD_LOG.debug("ReplCopyTask:\trcwork"); String distCpDoAsUser = conf.getVar(HiveConf.ConfVars.HIVE_DISTCP_DOAS_USER); rcwork.setDistCpDoAsUser(distCpDoAsUser); copyTask = TaskFactory.get(rcwork, conf); } else { - LOG.debug("ReplCopyTask:\tcwork"); + LOAD_LOG.debug("ReplCopyTask:\tcwork"); copyTask = TaskFactory.get(new CopyWork(srcPath, dstPath, false), conf); } return copyTask; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/StatsTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/StatsTask.java index 393d21dd0c..9af1e36c42 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/StatsTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/StatsTask.java @@ -28,8 +28,6 @@ import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.conf.HiveConf.ConfVars; import org.apache.hadoop.hive.ql.Context; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.apache.hadoop.hive.ql.TaskQueue; import org.apache.hadoop.hive.ql.QueryPlan; import org.apache.hadoop.hive.ql.QueryState; @@ -54,7 +52,6 @@ public class StatsTask extends Task implements Serializable { private static final long serialVersionUID = 1L; - private static transient final Logger LOG = LoggerFactory.getLogger(StatsTask.class); public StatsTask() { super(); @@ -147,12 +144,4 @@ protected void receiveFeed(FeedType feedType, Object feedValue) { dpPartSpecs = (Collection) feedValue; } } - - public static ExecutorService newThreadPool(HiveConf conf) { - int numThreads = HiveConf.getIntVar(conf, ConfVars.HIVE_STATS_GATHER_NUM_THREADS); - - ExecutorService executor = Executors.newFixedThreadPool(numThreads, new ThreadFactoryBuilder().setDaemon(true).setNameFormat("StatsNoJobTask-Thread-%d").build()); - LOG.info("Initialized threadpool for stats computation with {} threads", numThreads); - return executor; - } } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java index 86a2964858..eaf27b408b 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java @@ -55,6 +55,9 @@ public abstract class Task implements Serializable, Node { private static final long serialVersionUID = 1L; + + protected final transient Logger LOG = LoggerFactory.getLogger(getClass()); + public transient HashMap taskCounters; public transient TaskHandle taskHandle; protected transient HiveConf conf; @@ -67,7 +70,6 @@ protected transient String jobID; protected Task backupTask; protected List> backupChildrenTasks = new ArrayList>(); - protected static transient Logger LOG = LoggerFactory.getLogger(Task.class); protected int taskTag; private boolean isLocalMode =false; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java index 46764d4e30..cd107aa0ef 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java @@ -38,7 +38,6 @@ import org.apache.hadoop.hive.ql.exec.SerializationUtilities; import org.apache.hadoop.hive.ql.log.LogDivertAppenderForTest; import org.apache.hadoop.mapreduce.MRJobConfig; -import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.filecache.DistributedCache; @@ -123,8 +122,6 @@ private transient boolean isShutdown = false; private transient boolean jobKilled = false; - protected static transient final Logger LOG = LoggerFactory.getLogger(ExecDriver.class); - private RunningJob rj; /** @@ -717,8 +714,7 @@ public static void main(String[] args) throws IOException, HiveException { setupChildLog4j(conf); } - Logger LOG = LoggerFactory.getLogger(ExecDriver.class.getName()); - LogHelper console = new LogHelper(LOG, isSilent); + final LogHelper console = new LogHelper(LoggerFactory.getLogger(ExecDriver.class.getName()), isSilent); if (planFileName == null) { console.printError("Must specify Plan File Name"); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java index 78e3fd57f4..51d9182e10 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java @@ -81,8 +81,6 @@ import org.apache.hadoop.util.ReflectionUtils; import org.apache.hive.common.util.HiveStringUtils; import org.apache.hive.common.util.StreamPrinter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * MapredLocalTask represents any local work (i.e.: client side work) that hive needs to @@ -99,10 +97,8 @@ private final Map fetchOperators = new HashMap(); protected HadoopJobExecHelper jobExecHelper; private JobConf job; - public static transient final Logger l4j = LoggerFactory.getLogger(MapredLocalTask.class); static final String HIVE_LOCAL_TASK_CHILD_OPTS_KEY = "HIVE_LOCAL_TASK_CHILD_OPTS"; public static MemoryMXBean memoryMXBean; - private static final Logger LOG = LoggerFactory.getLogger(MapredLocalTask.class); // not sure we need this exec context; but all the operators in the work // will pass this context throught @@ -402,7 +398,7 @@ public int executeInProcess() { message = "Hive Runtime Error: Map local work failed"; retVal = 2; } - l4j.error(message, throwable); + LOG.error(message, throwable); console.printError(message, HiveStringUtils.stringifyException(throwable)); return retVal; } @@ -490,7 +486,7 @@ private void initializeOperators(Map fetchOpJobConfMap) FetchOperator fetchOp = new FetchOperator(entry.getValue(), jobClone); fetchOpJobConfMap.put(fetchOp, jobClone); fetchOperators.put(entry.getKey(), fetchOp); - l4j.info("fetchoperator for " + entry.getKey() + " created"); + LOG.info("fetchoperator for " + entry.getKey() + " created"); } // initialize all forward operator for (Map.Entry entry : fetchOperators.entrySet()) { @@ -510,7 +506,7 @@ private void initializeOperators(Map fetchOpJobConfMap) // initialize the forward operator ObjectInspector objectInspector = fetchOp.getOutputObjectInspector(); forwardOp.initialize(jobConf, new ObjectInspector[] {objectInspector}); - l4j.info("fetchoperator for " + entry.getKey() + " initialized"); + LOG.info("fetchoperator for " + entry.getKey() + " initialized"); } } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/ExternalTableCopyTaskBuilder.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/ExternalTableCopyTaskBuilder.java index 5c5543cfd9..e14a44c579 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/ExternalTableCopyTaskBuilder.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/ExternalTableCopyTaskBuilder.java @@ -69,7 +69,6 @@ } public static class DirCopyTask extends Task implements Serializable { - private static final Logger LOG = LoggerFactory.getLogger(DirCopyTask.class); private static final int MAX_COPY_RETRY = 5; private boolean createAndSetPathOwner(Path destPath, Path sourcePath) throws IOException { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java index f401b4dcea..10b8549a33 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java @@ -40,8 +40,6 @@ import org.apache.hadoop.hive.ql.exec.spark.status.SparkStage; import org.apache.hive.spark.counter.SparkCounter; import org.apache.hive.spark.counter.SparkCounters; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.ErrorMsg; @@ -79,8 +77,7 @@ public class SparkTask extends Task { private static final String CLASS_NAME = SparkTask.class.getName(); - private static final Logger LOG = LoggerFactory.getLogger(CLASS_NAME); - private static final LogHelper console = new LogHelper(LOG); + private transient final LogHelper console; private PerfLogger perfLogger; private static final long serialVersionUID = 1L; // The id of the actual Spark job @@ -99,6 +96,11 @@ private transient boolean isShutdown = false; private transient boolean jobKilled = false; + public SparkTask() { + super(); + this.console = new LogHelper(LOG); + } + @Override public int execute() { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java index 0bacb054e8..e57952c667 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java @@ -84,8 +84,6 @@ import org.apache.tez.dag.api.client.VertexStatus; import org.apache.tez.runtime.library.api.TezRuntimeConfiguration; import org.json.JSONObject; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import com.google.common.annotations.VisibleForTesting; @@ -99,7 +97,6 @@ public class TezTask extends Task { private static final String CLASS_NAME = TezTask.class.getName(); - private static transient Logger LOG = LoggerFactory.getLogger(CLASS_NAME); private final PerfLogger perfLogger = SessionState.getPerfLogger(); private static final String TEZ_MEMORY_RESERVE_FRACTION = "tez.task.scale.memory.reserve-fraction"; @@ -521,7 +518,7 @@ DAG build(JobConf conf, TezWork work, Path scratchDir, Context ctx, return dag; } - private static void setAccessControlsForCurrentUser(DAG dag, String queryId, + private void setAccessControlsForCurrentUser(DAG dag, String queryId, Configuration conf) throws IOException { String user = SessionState.getUserFromAuthenticator(); @@ -628,7 +625,7 @@ int close(TezWork work, int rc, DAGClient dagClient) { * Close DagClient, log warning if it throws any exception. * We don't want to fail query if that function fails. */ - private static void closeDagClientWithoutEx(DAGClient dagClient) { + private void closeDagClientWithoutEx(DAGClient dagClient) { try { dagClient.close(); } catch (Exception e) { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java index 53b3065a88..7b3d0fd884 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.apache.hadoop.fs.FileStatus; @@ -32,10 +33,10 @@ import org.apache.hadoop.hive.common.HiveStatsUtils; import org.apache.hadoop.hive.common.StatsSetupConst; import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.HiveConf.ConfVars; import org.apache.hadoop.hive.metastore.api.EnvironmentContext; import org.apache.hadoop.hive.metastore.api.InvalidOperationException; import org.apache.hadoop.hive.ql.CompilationOpContext; -import org.apache.hadoop.hive.ql.exec.StatsTask; import org.apache.hadoop.hive.ql.exec.Utilities; import org.apache.hadoop.hive.ql.io.AcidUtils; import org.apache.hadoop.hive.ql.io.StatsProvidingRecordReader; @@ -64,6 +65,7 @@ import com.google.common.collect.ImmutableListMultimap; import com.google.common.collect.Lists; import com.google.common.collect.Multimaps; +import com.google.common.util.concurrent.ThreadFactoryBuilder; /** * StatsNoJobTask is used in cases where stats collection is the only task for the given query (no @@ -106,7 +108,11 @@ public int process(Hive db, Table tbl) throws Exception { LOG.info("Executing stats (no job) task"); - ExecutorService threadPool = StatsTask.newThreadPool(conf); + final int numThreads = HiveConf.getIntVar(conf, ConfVars.HIVE_STATS_GATHER_NUM_THREADS); + final ExecutorService threadPool = Executors.newFixedThreadPool(numThreads, + new ThreadFactoryBuilder().setDaemon(true).setNameFormat("StatsNoJobTask-Thread-%d").build()); + + LOG.info("Initialized threadpool for stats computation with {} threads", numThreads); return aggregateStats(threadPool, db); }