• Hive-源码分析一条hql的执行过程

    一、源码下载

     下面是hive官方源码下载地址，我下载的是hive-3.1.3，那就一起来看下吧

    https://dlcdn.apache.org/hive/hive-3.1.3/apache-hive-3.1.3-src.tar.gz

    二、上下文

    <Hive-源码带你看hive命令背后都做了什么>博客中已经讲到了hive命令执行后会一直循环处理控制台输入的hql，下面就来继续分析下一条hql的执行过程，我们先看官网给的路径，然后再从源码开始捋。

    三、官网说明

    Design - Apache Hive - Apache Software Foundation

    图中还展示了一个典型的查询是如何在系统中流动的，这里我们先看普通的查询

    1、UI调用驱动程序的执行接口

    2、驱动程序为查询创建会话句柄，并将查询发送给编译器以生成执行计划

    3、4、编译器从元存储中获取必要的元数据

    5、利用元数据对查询树中的表达式进行类型检查，并根据查询谓词修剪分区。编译器生成计划，计划是阶段的DAG，每个阶段要么是Map/Reduce作业，要么是元数据操作，要么是HDFS上的操作。对于Map/Reduce阶段，计划包含map运算符树（在MapTask上执行的运算符树）和reduce运算符树（用于需要ReduceTask的操作）。

    6、6.1、6.2、6.3：执行引擎将这些阶段提交给适当的组件，

    四、源码分析

    <Hive-源码带你看hive命令背后都做了什么>博客中已经讲到了CliDriver.executeDriver()，我们从其中的processLine()开始捋

    1、processLine

    1.   /**
    2.    * 处理一行分号分隔的命令   *
    3.    * @param line
    4.    *          要处理的命令 也就是一条hql
    5.    * @param allowInterrupting
    6.    *          当为true时，函数将通过中断处理并返回-1来处理SIG_INT（Ctrl+C）
    7.    *
    8.    * @return 如果一切正常 返回 0 
    9.    */
    10.   public int processLine(String line, boolean allowInterrupting) {
    11.     SignalHandler oldSignal = null;
    12.     Signal interruptSignal = null;
    13.  
    14.     //如果是解析从控制台来的hql，allowInterrupting = true
    15.     if (allowInterrupting) {
    16.       //请记住在我们开始行处理时正在运行的所有线程。处理此行时挂起自定义Ctrl+C处理程序
    17.       //中断保留现场
    18.       interruptSignal = new Signal("INT");
    19.       oldSignal = Signal.handle(interruptSignal, new SignalHandler() {
    20.         private boolean interruptRequested;
    21.  
    22.         @Override
    23.         public void handle(Signal signal) {
    24.           boolean initialRequest = !interruptRequested;
    25.           interruptRequested = true;
    26.  
    27.           //在第二个ctrl+c上杀死VM
    28.           if (!initialRequest) {
    29.             console.printInfo("Exiting the JVM");
    30.             System.exit(127);
    31.           }
    32.  
    33.           //中断CLI线程以停止当前语句并返回提示，还确实，下方给出了截图
    34.           console.printInfo("Interrupting... Be patient, this might take some time.");
    35.           console.printInfo("Press Ctrl+C again to kill JVM");
    36.  
    37.           //首先，终止所有正在运行的MR作业
    38.           HadoopJobExecHelper.killRunningJobs();
    39.           TezJobExecHelper.killRunningJobs();
    40.           HiveInterruptUtils.interrupt();
    41.         }
    42.       });
    43.     }
    44.  
    45.     try {
    46.       int lastRet = 0, ret = 0;
    47.  
    48.       //我们不能直接使用“split”函数，因为可能会引用“;” 比如拼接字符串中有 “\\;”
    49.       //将hql按照字符一个一个处理，遇到 “;” 就会将前面的处理成一个hql 放入 commands 
    50.       List commands = splitSemiColon(line);
    51.  
    52.       String command = "";
    53.       //循环执行用户一次输入的多条hql
    54.       for (String oneCmd : commands) {
    55.  
    56.         if (StringUtils.endsWith(oneCmd, "\\")) {
    57.           command += StringUtils.chop(oneCmd) + ";";
    58.           continue;
    59.         } else {
    60.           command += oneCmd;
    61.         }
    62.         if (StringUtils.isBlank(command)) {
    63.           continue;
    64.         }
    65.         //接下来我们看processCmd方法中都做了什么
    66.         ret = processCmd(command);
    67.         command = "";
    68.         lastRet = ret;
    69.         boolean ignoreErrors = HiveConf.getBoolVar(conf, HiveConf.ConfVars.CLIIGNOREERRORS);
    70.         if (ret != 0 && !ignoreErrors) {
    71.           return ret;
    72.         }
    73.       }
    74.       return lastRet;
    75.     } finally {
    76.       // Once we are done processing the line, restore the old handler
    77.       if (oldSignal != null && interruptSignal != null) {
    78.         Signal.handle(interruptSignal, oldSignal);
    79.       }
    80.     }
    81.   }

    确实如源码中所写，当hql执行时如果按了ctrl+c 会有退出且给出这样的提示

    2、processCmd

    1.   public int processCmd(String cmd) {
    2.     CliSessionState ss = (CliSessionState) SessionState.get();
    3.     ss.setLastCommand(cmd);
    4.  
    5.     ss.updateThreadName();
    6.  
    7.     //刷新打印流，使其不包括上一个命令的输出
    8.     ss.err.flush();
    9.     //从sql语句中剥离注释，跟踪语句何时包含字符串文字。并去掉头尾空白符（只有头尾哟）
    10.     String cmd_trimmed = HiveStringUtils.removeComments(cmd).trim();
    11.     //将去掉注释和首尾空白的hql按照 "\\s+" 分割成 tokens 字符串数组 
    12.     // "\\s+" 等价于 [\f\r\t\v] 
    13.     //比如现在 tokens 就是{“select” ，“*” , “from” ,“ods.test” , "where" "dt='20240309'"}
    14.     String[] tokens = tokenizeCmd(cmd_trimmed);
    15.     int ret = 0;
    16.  
    17.     //如果用户输入的是 quit 或 exit 直接退出
    18.     if (cmd_trimmed.toLowerCase().equals("quit") || cmd_trimmed.toLowerCase().equals("exit")) {
    19.  
    20.       //如果我们已经走到了这一步——要么前面的命令都成功了，
    21.       //要么这是命令行。无论哪种情况，这都算作成功运行
    22.       ss.close();
    23.       System.exit(0);
    24.     //如果 hql 第一个字符串是 source
    25.     } else if (tokens[0].equalsIgnoreCase("source")) {
    26.       //获取 source 后的hql字符串
    27.       String cmd_1 = getFirstCmd(cmd_trimmed, tokens[0].length());
    28.       cmd_1 = new VariableSubstitution(new HiveVariableSource() {
    29.         @Override
    30.         public Map getHiveVariable() {
    31.           return SessionState.get().getHiveVariables();
    32.         }
    33.       }).substitute(ss.getConf(), cmd_1);
    34.  
    35.       File sourceFile = new File(cmd_1);
    36.       if (! sourceFile.isFile()){
    37.         console.printError("File: "+ cmd_1 + " is not a file.");
    38.         ret = 1;
    39.       } else {
    40.         try {
    41.           ret = processFile(cmd_1);
    42.         } catch (IOException e) {
    43.           console.printError("Failed processing file "+ cmd_1 +" "+ e.getLocalizedMessage(),
    44.             stringifyException(e));
    45.           ret = 1;
    46.         }
    47.       }
    48.     } else if (cmd_trimmed.startsWith("!")) {
    49.       // 对于shell命令，请使用unstretch命令
    50.       //可以在hive客户端输入 ! sh your_script.sh 执行你的脚本
    51.       String shell_cmd = cmd.trim().substring(1);
    52.       shell_cmd = new VariableSubstitution(new HiveVariableSource() {
    53.         @Override
    54.         public Map getHiveVariable() {
    55.           return SessionState.get().getHiveVariables();
    56.         }
    57.       }).substitute(ss.getConf(), shell_cmd);
    58.  
    59.       // shell_cmd = "/bin/bash -c \'" + shell_cmd + "\'";
    60.       try {
    61.         ShellCmdExecutor executor = new ShellCmdExecutor(shell_cmd, ss.out, ss.err);
    62.         ret = executor.execute();
    63.         if (ret != 0) {
    64.           console.printError("Command failed with exit code = " + ret);
    65.         }
    66.       } catch (Exception e) {
    67.         console.printError("Exception raised from Shell command " + e.getLocalizedMessage(),
    68.             stringifyException(e));
    69.         ret = 1;
    70.       }
    71.     }  else { //本地方式
    72.       try {
    73.         //获取执行hql的驱动程序，这个我们详细来看下
    74.         try (CommandProcessor proc = CommandProcessorFactory.get(tokens, (HiveConf) conf)) {
    75.           if (proc instanceof IDriver) {
    76.             //让驱动程序使用sql解析器剥离注释
    77.             ret = processLocalCmd(cmd, proc, ss);
    78.           } else {
    79.             //这里是直接使用剥离完注释的sql，我们看这里
    80.             ret = processLocalCmd(cmd_trimmed, proc, ss);
    81.           }
    82.         }
    83.       } catch (SQLException e) {
    84.         console.printError("Failed processing command " + tokens[0] + " " + e.getLocalizedMessage(),
    85.           org.apache.hadoop.util.StringUtils.stringifyException(e));
    86.         ret = 1;
    87.       }
    88.       catch (Exception e) {
    89.         throw new RuntimeException(e);
    90.       }
    91.     }
    92.  
    93.     ss.resetThreadName();
    94.     return ret;
    95.   }

    3、获取执行hql的驱动程序

    顺着第2步看这个类CommandProcessorFactory

    1.   public static CommandProcessor get(String[] cmd, @Nonnull HiveConf conf) throws SQLException {
    2.     CommandProcessor result = getForHiveCommand(cmd, conf);
    3.     if (result != null) {
    4.       return result;
    5.     }
    6.     if (isBlank(cmd[0])) {
    7.       return null;
    8.     } else {
    9.       //如果不是llap开头的hql都会走这
    10.       //为客户端构建一个驱动程序
    11.       return DriverFactory.newDriver(conf);
    12.     }
    13.   }
    14.  
    15.   public static CommandProcessor getForHiveCommand(String[] cmd, HiveConf conf)
    16.     throws SQLException {
    17.     return getForHiveCommandInternal(cmd, conf, false);
    18.   }
    19.  
    20. public static CommandProcessor getForHiveCommandInternal(String[] cmd, HiveConf conf,
    21.                                                            boolean testOnly)
    22.     throws SQLException {
    23.     //这部分是关键，在HiveCommand中，我们看下
    24.     HiveCommand hiveCommand = HiveCommand.find(cmd, testOnly);
    25.     if (hiveCommand == null || isBlank(cmd[0])) {
    26.       return null;
    27.     }
    28.     if (conf == null) {
    29.       conf = new HiveConf();
    30.     }
    31.     Set availableCommands = new HashSet();
    32.     for (String availableCommand : conf.getVar(HiveConf.ConfVars.HIVE_SECURITY_COMMAND_WHITELIST)
    33.       .split(",")) {
    34.       availableCommands.add(availableCommand.toLowerCase().trim());
    35.     }
    36.     if (!availableCommands.contains(cmd[0].trim().toLowerCase())) {
    37.       throw new SQLException("Insufficient privileges to execute " + cmd[0], "42000");
    38.     }
    39.     if (cmd.length > 1 && "reload".equalsIgnoreCase(cmd[0])
    40.       && "function".equalsIgnoreCase(cmd[1])) {
    41.       // special handling for SQL "reload function"
    42.       return null;
    43.     }
    44.     switch (hiveCommand) {
    45.       case SET:
    46.         return new SetProcessor();
    47.       case RESET:
    48.         return new ResetProcessor();
    49.       case DFS:
    50.         SessionState ss = SessionState.get();
    51.         return new DfsProcessor(ss.getConf());
    52.       case ADD:
    53.         return new AddResourceProcessor();
    54.       case LIST:
    55.         return new ListResourceProcessor();
    56.       case LLAP_CLUSTER:
    57.         return new LlapClusterResourceProcessor();
    58.       case LLAP_CACHE:
    59.         return new LlapCacheResourceProcessor();
    60.       case DELETE:
    61.         return new DeleteResourceProcessor();
    62.       case COMPILE:
    63.         return new CompileProcessor();
    64.       case RELOAD:
    65.         return new ReloadProcessor();
    66.       case CRYPTO:
    67.         try {
    68.           return new CryptoProcessor(SessionState.get().getHdfsEncryptionShim(), conf);
    69.         } catch (HiveException e) {
    70.           throw new SQLException("Fail to start the command processor due to the exception: ", e);
    71.         }
    72.       default:
    73.         throw new AssertionError("Unknown HiveCommand " + hiveCommand);
    74.     }
    75.   }

    HiveCommand是非SQL语句，例如设置属性或添加资源。

    1. //可以看出正常情况下只会返回  LLAP_CLUSTER 和 LLAP_CACHE
    2. public static HiveCommand find(String[] command, boolean findOnlyForTesting) {
    3.     if (null == command){
    4.       return null;
    5.     }
    6.     //解析第一个hql字符串，比如 select 、 delete 、update 、set 等等
    7.     String cmd = command[0];
    8.     if (cmd != null) {
    9.       /转成大写 SELECT 、 DELETE 、UPDATE 、SET 等等
    10.       cmd = cmd.trim().toUpperCase();
    11.       if (command.length > 1 && "role".equalsIgnoreCase(command[1])) {
    12.         //对 "set role r1" 语句的特殊处理
    13.         return null;
    14.       } else if(command.length > 1 && "from".equalsIgnoreCase(command[1])) {
    15.         //对 "delete from  where..." 语句特殊处理
    16.         return null;
    17.       } else if(command.length > 1 && "set".equalsIgnoreCase(command[0]) && "autocommit".equalsIgnoreCase(command[1])) {
    18.         return null;//不希望set autocommit true|false与set hive.foo.bar混合......
    19.       } else if (command.length > 1 && "llap".equalsIgnoreCase(command[0])) {
    20.         return getLlapSubCommand(command);
    21.       } else if (COMMANDS.contains(cmd)) {
    22.         HiveCommand hiveCommand = HiveCommand.valueOf(cmd);
    23.  
    24.         if (findOnlyForTesting == hiveCommand.isOnlyForTesting()) {
    25.           return hiveCommand;
    26.         }
    27.  
    28.         return null;
    29.       }
    30.     }
    31.     return null;
    32.   }
    33.  
    34.   private static HiveCommand getLlapSubCommand(final String[] command) {
    35.     if ("cluster".equalsIgnoreCase(command[1])) {
    36.       return LLAP_CLUSTER;
    37.     } else if ("cache".equalsIgnoreCase(command[1])) {
    38.       return LLAP_CACHE;
    39.     } else {
    40.       return null;
    41.     }
    42.   }
      43. 如果不是llap开头的hql都会走这 return DriverFactory.newDriver(conf);
      1.   public static IDriver newDriver(QueryState queryState, String userName, QueryInfo queryInfo) {
      2.     //获取配置中 hive.query.reexecution.enabled 的属性值 默认 true
      3.     //解释：启用查询重新执行
      4.     boolean enabled = queryState.getConf().getBoolVar(ConfVars.HIVE_QUERY_REEXECUTION_ENABLED);
      5.     if (!enabled) {
      6.       //如果没有开启则返回Driver
      7.       return new Driver(queryState, userName, queryInfo);
      8.     }
      9.  
      10.     //获取配置中 hive.query.reexecution.strategies 的属性值 默认值为 overlay,reoptimize
      11.     //解释：可以使用逗号分隔的插件列表：
      12.     //overlay：hiveconf子树“reexec.overlay”用作执行出错时的覆盖
      13.     //reoptimize：在执行期间收集运算符统计信息，并在失败后重新编译查询
      14.     String strategies = queryState.getConf().getVar(ConfVars.HIVE_QUERY_REEXECUTION_STRATEGIES);
      15.     strategies = Strings.nullToEmpty(strategies).trim().toLowerCase();
      16.     ArrayList plugins = new ArrayList<>();
      17.     for (String string : strategies.split(",")) {
      18.       if (string.trim().isEmpty()) {
      19.         continue;
      20.       }
      21.       plugins.add(buildReExecPlugin(string));
      22.     }
      23.     //默认返回ReExecDriver
      24.     //覆盖IDriver接口，处理查询的重新执行；并向底层的重新执行插件提出了明确的问题。
      25.     return new ReExecDriver(queryState, userName, queryInfo, plugins);
      26.   }
      4、processLocalCmd
      1.  int processLocalCmd(String cmd, CommandProcessor proc, CliSessionState ss) {
      2.     //获取hive-site.xml中的hive.cli.print.escape.crlf属性值，默认为false
      3.     //解释：是否将行输出中的回车和换行打印为转义符\r\n
      4.     boolean escapeCRLF = HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_CLI_PRINT_ESCAPE_CRLF);
      5.     int ret = 0;
      6.  
      7.     if (proc != null) {
      8.       //从第3步已经知晓，默认会走这一步
      9.       if (proc instanceof IDriver) {
      10.         //强制先转成IDriver 
      11.         IDriver qp = (IDriver) proc;
      12.         PrintStream out = ss.out;
      13.         long start = System.currentTimeMillis();
      14.         if (ss.getIsVerbose()) {
      15.           out.println(cmd);
      16.         }
      17.         //这里调用的时IDriver.run() 我们详细看下
      18.         ret = qp.run(cmd).getResponseCode();
      19.         if (ret != 0) {
      20.           qp.close();
      21.           return ret;
      22.         }
      23.  
      24.         //查询已运行捕获时间
      25.         long end = System.currentTimeMillis();
      26.         double timeTaken = (end - start) / 1000.0;
      27.  
      28.         ArrayList res = new ArrayList();
      29.  
      30.         printHeader(qp, out);
      31.  
      32.         //打印结果
      33.         int counter = 0;
      34.         try {
      35.           if (out instanceof FetchConverter) {
      36.             ((FetchConverter) out).fetchStarted();
      37.           }
      38.           while (qp.getResults(res)) {
      39.             for (String r : res) {
      40.                   if (escapeCRLF) {
      41.                     r = EscapeCRLFHelper.escapeCRLF(r);
      42.                   }
      43.               out.println(r);
      44.             }
      45.             counter += res.size();
      46.             res.clear();
      47.             if (out.checkError()) {
      48.               break;
      49.             }
      50.           }
      51.         } catch (IOException e) {
      52.           console.printError("Failed with exception " + e.getClass().getName() + ":" + e.getMessage(),
      53.               "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
      54.           ret = 1;
      55.         }
      56.  
      57.         qp.close();
      58.  
      59.         if (out instanceof FetchConverter) {
      60.           ((FetchConverter) out).fetchFinished();
      61.         }
      62.  
      63.         console.printInfo(
      64.             "Time taken: " + timeTaken + " seconds" + (counter == 0 ? "" : ", Fetched: " + counter + " row(s)"));
      65.       } else {
      66.         String firstToken = tokenizeCmd(cmd.trim())[0];
      67.         String cmd_1 = getFirstCmd(cmd.trim(), firstToken.length());
      68.  
      69.         if (ss.getIsVerbose()) {
      70.           ss.out.println(firstToken + " " + cmd_1);
      71.         }
      72.         CommandProcessorResponse res = proc.run(cmd_1);
      73.         if (res.getResponseCode() != 0) {
      74.           ss.out
      75.               .println("Query returned non-zero code: " + res.getResponseCode() + ", cause: " + res.getErrorMessage());
      76.         }
      77.         if (res.getConsoleMessages() != null) {
      78.           for (String consoleMsg : res.getConsoleMessages()) {
      79.             console.printInfo(consoleMsg);
      80.           }
      81.         }
      82.         ret = res.getResponseCode();
      83.       }
      84.     }
      85.  
      86.     return ret;
      87.   }
      5、ReExecDriver
      1.   public CommandProcessorResponse run(String command) {
      2.     CommandProcessorResponse r0 = compileAndRespond(command);
      3.     if (r0.getResponseCode() != 0) {
      4.       return r0;
      5.     }
      6.     return run();
      7.   }
      8.  
      9.   public CommandProcessorResponse compileAndRespond(String statement) {
      10.     currentQuery = statement;
      11.     //coreDriver就是Driver 我们去Driver详细看下这个逻辑
      12.     return coreDriver.compileAndRespond(statement);
      13.   }
      14.  
      15.   public CommandProcessorResponse run() {
      16.     executionIndex = 0;
      17.     int maxExecutuions = 1 + coreDriver.getConf().getIntVar(ConfVars.HIVE_QUERY_MAX_REEXECUTION_COUNT);
      18.  
      19.  
      20.     while (true) {
      21.       executionIndex++;
      22.       for (IReExecutionPlugin p : plugins) {
      23.         p.beforeExecute(executionIndex, explainReOptimization);
      24.       }
      25.       coreDriver.getContext().setExecutionIndex(executionIndex);
      26.       LOG.info("Execution #{} of query", executionIndex);
      27.       CommandProcessorResponse cpr = coreDriver.run();
      28.  
      29.       PlanMapper oldPlanMapper = coreDriver.getPlanMapper();
      30.       afterExecute(oldPlanMapper, cpr.getResponseCode() == 0);
      31.  
      32.       boolean shouldReExecute = explainReOptimization && executionIndex==1;
      33.       shouldReExecute |= cpr.getResponseCode() != 0 && shouldReExecute();
      34.  
      35.       if (executionIndex >= maxExecutuions || !shouldReExecute) {
      36.         return cpr;
      37.       }
      38.       LOG.info("Preparing to re-execute query");
      39.       prepareToReExecute();
      40.       CommandProcessorResponse compile_resp = coreDriver.compileAndRespond(currentQuery);
      41.       if (compile_resp.failed()) {
      42.         LOG.error("Recompilation of the query failed; this is unexpected.");
      43.         // FIXME: somehow place pointers that re-execution compilation have failed; the query have been successfully compiled before?
      44.         return compile_resp;
      45.       }
      46.  
      47.       PlanMapper newPlanMapper = coreDriver.getPlanMapper();
      48.       if (!explainReOptimization && !shouldReExecuteAfterCompile(oldPlanMapper, newPlanMapper)) {
      49.         LOG.info("re-running the query would probably not yield better results; returning with last error");
      50.         // FIXME: retain old error; or create a new one?
      51.         return cpr;
      52.       }
      53.     }
      54.   }
      5.1、Driver
      1.   public CommandProcessorResponse compileAndRespond(String command, boolean cleanupTxnList) {
      2.     try {
      3.       compileInternal(command, false);
      4.       return createProcessorResponse(0);
      5.     } catch (CommandProcessorResponse e) {
      6.       return e;
      7.     } finally {
      8.       if (cleanupTxnList) {
      9.         //使用此命令编译的查询可能会生成有效的txn列表，因此我们需要重置它
      10.         conf.unset(ValidTxnList.VALID_TXNS_KEY);
      11.       }
      12.     }
      13.   }
      14.  
      15.   private void compileInternal(String command, boolean deferClose) throws CommandProcessorResponse {
      16.     
      17.     //......省略......    
      18.  
      19.     try {
      20.       //deferClose表示进程中断时是否应推迟关闭/销毁，
      21.       //如果在另一个方法（如runInternal）内调用编译，
      22.       //则应将其设置为true，runInternal将关闭推迟到该方法中调用的。
      23.       //我们详细看下
      24.       compile(command, true, deferClose);
      25.     } catch (CommandProcessorResponse cpr) {
      26.  
      27.         //......省略......  
      28.  
      29.     } finally {
      30.       compileLock.unlock();
      31.     }
      32.  
      33.     //......省略......  
      34.   }
      35.  
      36.  
      37. private void compile(String command, boolean resetTaskIds, boolean deferClose) throws CommandProcessorResponse {
      38.     
      39.      //......省略...... 
      40.  
      41.     command = new VariableSubstitution(new HiveVariableSource() {
      42.       @Override
      43.       public Map getHiveVariable() {
      44.         return SessionState.get().getHiveVariables();
      45.       }
      46.     }).substitute(conf, command);
      47.  
      48.     String queryStr = command;
      49.  
      50.     try {
      51.       //应编辑命令以避免记录敏感数据
      52.       queryStr = HookUtils.redactLogString(conf, command);
      53.     } catch (Exception e) {
      54.       LOG.warn("WARNING! Query command could not be redacted." + e);
      55.     }
      56.  
      57.     checkInterrupted("at beginning of compilation.", null, null);
      58.  
      59.     if (ctx != null && ctx.getExplainAnalyze() != AnalyzeState.RUNNING) {
      60.       //在编译新查询之前关闭现有的ctx-etc，但不要破坏驱动程序
      61.       closeInProcess(false);
      62.     }
      63.  
      64.     if (resetTaskIds) {
      65.       TaskFactory.resetId();
      66.     }
      67.  
      68.     LockedDriverState.setLockedDriverState(lDrvState);
      69.  
      70.     //获取查询id 正在执行的查询的ID（每个会话可能有多个
      71.     String queryId = queryState.getQueryId();
      72.  
      73.     if (ctx != null) {
      74.       setTriggerContext(queryId);
      75.     }
      76.     //保存一些信息以供webUI在计划释放后使用
      77.     this.queryDisplay.setQueryStr(queryStr);
      78.     this.queryDisplay.setQueryId(queryId);
      79.  
      80.     //正在编译这条 hql 
      81.     LOG.info("Compiling command(queryId=" + queryId + "): " + queryStr);
      82.  
      83.     conf.setQueryString(queryStr);
      84.     //FIXME:副作用将把最后一个查询集留在会话级别
      85.     if (SessionState.get() != null) {
      86.       SessionState.get().getConf().setQueryString(queryStr);
      87.       SessionState.get().setupQueryCurrentTimestamp();
      88.     }
      89.  
      90.     //查询编译过程中是否发生任何错误。用于查询生存期挂钩。
      91.     boolean compileError = false;
      92.     boolean parseError = false;
      93.  
      94.     try {
      95.  
      96.       //初始化事务管理器。这必须在调用解析（analyze）之前完成。
      97.       if (initTxnMgr != null) {
      98.         queryTxnMgr = initTxnMgr;
      99.       } else {
      100.         queryTxnMgr = SessionState.get().initTxnMgr(conf);
      101.       }
      102.       if (queryTxnMgr instanceof Configurable) {
      103.         ((Configurable) queryTxnMgr).setConf(conf);
      104.       }
      105.       queryState.setTxnManager(queryTxnMgr);
      106.  
      107.       //如果用户Ctrl-C两次杀死Hive CLI JVM，如果多次调用compile，
      108.       //我们希望释放锁，请清除旧的shutdownhook
      109.       ShutdownHookManager.removeShutdownHook(shutdownRunner);
      110.       final HiveTxnManager txnMgr = queryTxnMgr;
      111.       shutdownRunner = new Runnable() {
      112.         @Override
      113.         public void run() {
      114.           try {
      115.             releaseLocksAndCommitOrRollback(false, txnMgr);
      116.           } catch (LockException e) {
      117.             LOG.warn("Exception when releasing locks in ShutdownHook for Driver: " +
      118.                 e.getMessage());
      119.           }
      120.         }
      121.       };
      122.       ShutdownHookManager.addShutdownHook(shutdownRunner, SHUTDOWN_HOOK_PRIORITY);
      123.  
      124.       //在解析和分析查询之前
      125.       checkInterrupted("before parsing and analysing the query", null, null);
      126.  
      127.       if (ctx == null) {
      128.         ctx = new Context(conf);
      129.         setTriggerContext(queryId);
      130.       }
      131.  
      132.       //设置此查询的事务管理器
      133.       ctx.setHiveTxnManager(queryTxnMgr);
      134.       ctx.setStatsSource(statsSource);
      135.       //设置hql
      136.       ctx.setCmd(command);
      137.       //退出时清理HDFS
      138.       ctx.setHDFSCleanup(true);
      139.  
      140.       perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.PARSE);
      141.  
      142.       //在查询进入解析阶段之前调用
      143.       hookRunner.runBeforeParseHook(command);
      144.  
      145.       ASTNode tree;
      146.       try {
      147.         //解析hql 这里先不展开讲，我们会单独拿一篇博客来研究
      148.         tree = ParseUtils.parse(command, ctx);    
      149.       } catch (ParseException e) {
      150.         parseError = true;
      151.         throw e;
      152.       } finally {
      153.         hookRunner.runAfterParseHook(command, parseError);
      154.       }
      155.       perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.PARSE);
      156.  
      157.       hookRunner.runBeforeCompileHook(command);
      158.       //清除CurrentFunctionsInUse 设置，以捕获SemanticAnalyzer发现正在使用的新函数集
      159.       SessionState.get().getCurrentFunctionsInUse().clear();
      160.       perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ANALYZE);
      161.  
      162.       //刷新元存储缓存。这确保了我们不会从在同一线程中运行的先前查询中拾取对象。
      163.       //这必须在我们获得语义分析器之后（即与元存储建立连接时），
      164.       //但在我们进行分析之前完成，因为此时我们需要访问对象。
      165.       Hive.get().getMSC().flushCache();
      166.  
      167.       backupContext = new Context(ctx);
      168.       boolean executeHooks = hookRunner.hasPreAnalyzeHooks();
      169.  
      170.       //Hive为HiveSemanticAnalyzerHook的实现提供的上下文信息
      171.       HiveSemanticAnalyzerHookContext hookCtx = new HiveSemanticAnalyzerHookContextImpl();
      172.       if (executeHooks) {
      173.         hookCtx.setConf(conf);
      174.         hookCtx.setUserName(userName);
      175.         hookCtx.setIpAddress(SessionState.get().getUserIpAddress());
      176.         hookCtx.setCommand(command);
      177.         hookCtx.setHiveOperation(queryState.getHiveOperation());
      178.  
      179.         //在Hive对语句执行自己的语义分析之前调用。实现可以检查语句AST，
      180.         //并通过抛出SemanticException来阻止其执行。它是可选地，
      181.         //它也可以扩充/重写AST，但必须生成一个与Hive自己的解析器直接返回的表单等效的表单。
      182.         //返回替换后的AST（通常与原始AST相同，除非必须替换整个树；不得为null）
      183.         tree =  hookRunner.runPreAnalyzeHooks(hookCtx, tree);
      184.       }
      185.  
      186.       //进行语义分析和计划生成
      187.       //这里会根据 tree的type获取不同的优化引擎，默认是CalcitePlanner
      188.       BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(queryState, tree);
      189.  
      190.       if (!retrial) {
      191.         openTransaction();
      192.         generateValidTxnList();
      193.       }
      194.  
      195.       //对hql转化后的tree进行解析，比如：语义分析 ，后面专门用一篇博客来研究
      196.       sem.analyze(tree, ctx);
      197.  
      198.       if (executeHooks) {
      199.         hookCtx.update(sem);
      200.         hookRunner.runPostAnalyzeHooks(hookCtx, sem.getAllRootTasks());
      201.       }
      202.  
      203.       /语义分析完成
      204.       LOG.info("Semantic Analysis Completed (retrial = {})", retrial);
      205.  
      206.       //检索有关查询的缓存使用情况的信息。
      207.       if (conf.getBoolVar(HiveConf.ConfVars.HIVE_QUERY_RESULTS_CACHE_ENABLED)) {
      208.         cacheUsage = sem.getCacheUsage();
      209.       }
      210.  
      211.       //验证计划
      212.       sem.validate();
      213.       perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ANALYZE);
      214.  
      215.       //分析查询后
      216.       checkInterrupted("after analyzing query.", null, null);
      217.  
      218.       //获取输出模式
      219.       schema = getSchema(sem, conf);
      220.       //制作查询计划
      221.       plan = new QueryPlan(queryStr, sem, perfLogger.getStartTime(PerfLogger.DRIVER_RUN), queryId,
      222.         queryState.getHiveOperation(), schema);
      223.  
      224.       //设置mapreduce工作流引擎id和name
      225.       conf.set("mapreduce.workflow.id", "hive_" + queryId);
      226.       conf.set("mapreduce.workflow.name", queryStr);
      227.  
      228.       //在此处初始化FetchTask
      229.       if (plan.getFetchTask() != null) {
      230.         plan.getFetchTask().initialize(queryState, plan, null, ctx.getOpContext());
      231.       }
      232.  
      233.       //进行授权检查
      234.       if (!sem.skipAuthorization() &&
      235.           HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_AUTHORIZATION_ENABLED)) {
      236.  
      237.         try {
      238.           perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.DO_AUTHORIZATION);
      239.           //具体会做以下操作
      240.           //    1、连接hive的元数据
      241.           //    2、设置输入输出
      242.           //    3、获取表和列的映射
      243.           //    4、添加正在使用的永久UDF
      244.           //    5、解析hql操作是对数据库、表、还是查询或者导入
      245.           //    6、如果是分区表，还要检查分区权限
      246.           //    7、通过表扫描运算符检查列授权
      247.           //    8、表授权检查
      248.           doAuthorization(queryState.getHiveOperation(), sem, command);
      249.         } catch (AuthorizationException authExp) {
      250.           console.printError("Authorization failed:" + authExp.getMessage()
      251.               + ". Use SHOW GRANT to get more details.");
      252.           errorMessage = authExp.getMessage();
      253.           SQLState = "42000";
      254.           throw createProcessorResponse(403);
      255.         } finally {
      256.           perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.DO_AUTHORIZATION);
      257.         }
      258.       }
      259.  
      260.       if (conf.getBoolVar(ConfVars.HIVE_LOG_EXPLAIN_OUTPUT)) {
      261.         String explainOutput = getExplainOutput(sem, plan, tree);
      262.         if (explainOutput != null) {
      263.           LOG.info("EXPLAIN output for queryid " + queryId + " : "
      264.             + explainOutput);
      265.           if (conf.isWebUiQueryInfoCacheEnabled()) {
      266.             //设置执行计划
      267.             queryDisplay.setExplainPlan(explainOutput);
      268.           }
      269.         }
      270.       }
      271.     } catch (CommandProcessorResponse cpr) {
      272.       throw cpr;
      273.     } catch (Exception e) {
      274.       checkInterrupted("during query compilation: " + e.getMessage(), null, null);
      275.  
      276.       compileError = true;
      277.       ErrorMsg error = ErrorMsg.getErrorMsg(e.getMessage());
      278.       errorMessage = "FAILED: " + e.getClass().getSimpleName();
      279.       if (error != ErrorMsg.GENERIC_ERROR) {
      280.         errorMessage += " [Error "  + error.getErrorCode()  + "]:";
      281.       }
      282.  
      283.       // HIVE-4889
      284.       if ((e instanceof IllegalArgumentException) && e.getMessage() == null && e.getCause() != null) {
      285.         errorMessage += " " + e.getCause().getMessage();
      286.       } else {
      287.         errorMessage += " " + e.getMessage();
      288.       }
      289.  
      290.       if (error == ErrorMsg.TXNMGR_NOT_ACID) {
      291.         errorMessage += ". Failed command: " + queryStr;
      292.       }
      293.  
      294.       SQLState = error.getSQLState();
      295.       downstreamError = e;
      296.       console.printError(errorMessage, "\n"
      297.           + org.apache.hadoop.util.StringUtils.stringifyException(e));
      298.       throw createProcessorResponse(error.getErrorCode());
      299.     } finally {
      300.       // 触发编译后挂钩。请注意，如果此处编译失败，则执行前/执行后挂钩将永远不会执行。
      301.       if (!parseError) {
      302.         try {
      303.           hookRunner.runAfterCompilationHook(command, compileError);
      304.         } catch (Exception e) {
      305.           LOG.warn("Failed when invoking query after-compilation hook.", e);
      306.         }
      307.       }
      308.  
      309.       double duration = perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.COMPILE)/1000.00;
      310.       ImmutableMap compileHMSTimings = dumpMetaCallTimingWithoutEx("compilation");
      311.       queryDisplay.setHmsTimings(QueryDisplay.Phase.COMPILATION, compileHMSTimings);
      312.  
      313.       boolean isInterrupted = lDrvState.isAborted();
      314.       if (isInterrupted && !deferClose) {
      315.         closeInProcess(true);
      316.       }
      317.       lDrvState.stateLock.lock();
      318.       try {
      319.         if (isInterrupted) {
      320.           lDrvState.driverState = deferClose ? DriverState.EXECUTING : DriverState.ERROR;
      321.         } else {
      322.           lDrvState.driverState = compileError ? DriverState.ERROR : DriverState.COMPILED;
      323.         }
      324.       } finally {
      325.         lDrvState.stateLock.unlock();
      326.       }
      327.  
      328.       if (isInterrupted) {
      329.         LOG.info("Compiling command(queryId=" + queryId + ") has been interrupted after " + duration + " seconds");
      330.       } else {
      331.         LOG.info("Completed compiling command(queryId=" + queryId + "); Time taken: " + duration + " seconds");
      332.       }
      333.     }
      334.   }
      5.2、ReExecDriver自身执行
      1.   public CommandProcessorResponse run() {
      2.     executionIndex = 0;
      3.     //获取配置文件中的 hive.query.reexecution.max.count 属性值，默认为 1
      4.     //解释：单个查询的最大重新执行次数
      5.     int maxExecutuions = 1 + coreDriver.getConf().getIntVar(ConfVars.HIVE_QUERY_MAX_REEXECUTION_COUNT);
      6.  
      7.  
      8.     while (true) {
      9.       executionIndex++;
      10.       //循环执行重新执行逻辑
      11.       for (IReExecutionPlugin p : plugins) {
      12.         //在执行查询之前调用
      13.         p.beforeExecute(executionIndex, explainReOptimization);
      14.       }
      15.       coreDriver.getContext().setExecutionIndex(executionIndex);
      16.       LOG.info("Execution #{} of query", executionIndex);
      17.       //还是会调用Driver ，但是和5.1调用的不一样，我们详细看看
      18.       CommandProcessorResponse cpr = coreDriver.run();
      19.  
      20.       PlanMapper oldPlanMapper = coreDriver.getPlanMapper();
      21.       afterExecute(oldPlanMapper, cpr.getResponseCode() == 0);
      22.  
      23.       boolean shouldReExecute = explainReOptimization && executionIndex==1;
      24.       shouldReExecute |= cpr.getResponseCode() != 0 && shouldReExecute();
      25.  
      26.       if (executionIndex >= maxExecutuions || !shouldReExecute) {
      27.         return cpr;
      28.       }
      29.       //正在准备重新执行查询
      30.       LOG.info("Preparing to re-execute query");
      31.       prepareToReExecute();
      32.       CommandProcessorResponse compile_resp = coreDriver.compileAndRespond(currentQuery);
      33.       if (compile_resp.failed()) {
      34.         LOG.error("Recompilation of the query failed; this is unexpected.");
      35.         return compile_resp;
      36.       }
      37.  
      38.       PlanMapper newPlanMapper = coreDriver.getPlanMapper();
      39.       if (!explainReOptimization && !shouldReExecuteAfterCompile(oldPlanMapper, newPlanMapper)) {
      40.         //重新运行查询可能不会产生更好的结果；返回最后一个错误
      41.         LOG.info("re-running the query would probably not yield better results; returning with last error");
      42.         return cpr;
      43.       }
      44.     }
      45.   }
      分析调用Driver的逻辑（和5.1不同）
      1. public CommandProcessorResponse run(String command, boolean alreadyCompiled) {
      2.  
      3.     try {
      4.       runInternal(command, alreadyCompiled);
      5.       return createProcessorResponse(0);
      6.     } catch (CommandProcessorResponse cpr) {
      7.  
      8.       //......省略......
      9.     }
      10.   }
      11.  
      12.  
      13. private void runInternal(String command, boolean alreadyCompiled) throws CommandProcessorResponse {
      14.     errorMessage = null;
      15.     SQLState = null;
      16.     downstreamError = null;
      17.     LockedDriverState.setLockedDriverState(lDrvState);
      18.  
      19.     lDrvState.stateLock.lock();
      20.     try {
      21.       if (alreadyCompiled) {
      22.         if (lDrvState.driverState == DriverState.COMPILED) {
      23.           //如果引擎是编译状态，现在修改成执行状态
      24.           lDrvState.driverState = DriverState.EXECUTING;
      25.         } else {
      26.           //失败：预编译的查询已被取消或关闭。
      27.           errorMessage = "FAILED: Precompiled query has been cancelled or closed.";
      28.           console.printError(errorMessage);
      29.           throw createProcessorResponse(12);
      30.         }
      31.       } else {
      32.         lDrvState.driverState = DriverState.COMPILING;
      33.       }
      34.     } finally {
      35.       lDrvState.stateLock.unlock();
      36.     }
      37.  
      38.     //一个标志，通过跟踪方法是否因错误而返回，帮助在finally块中设置正确的驱动器状态。
      39.     boolean isFinishedWithError = true;
      40.     try {
      41.       //Hive向HiveDriverRunHook的实现提供的上下文信息
      42.       HiveDriverRunHookContext hookContext = new HiveDriverRunHookContextImpl(conf,
      43.           alreadyCompiled ? ctx.getCmd() : command);
      44.         //获取所有驱动程序运行挂钩并预执行它们
      45.       try {
      46.         hookRunner.runPreDriverHooks(hookContext);
      47.       } catch (Exception e) {
      48.         errorMessage = "FAILED: Hive Internal Error: " + Utilities.getNameMessage(e);
      49.         SQLState = ErrorMsg.findSQLState(e.getMessage());
      50.         downstreamError = e;
      51.         console.printError(errorMessage + "\n"
      52.             + org.apache.hadoop.util.StringUtils.stringifyException(e));
      53.         throw createProcessorResponse(12);
      54.       }
      55.  
      56.       PerfLogger perfLogger = null;
      57.  
      58.       //如果还没有编译
      59.       if (!alreadyCompiled) {
      60.         //内部编译将自动重置性能记录器
      61.         compileInternal(command, true);
      62.         //然后我们继续使用这个性能记录器
      63.         perfLogger = SessionState.getPerfLogger();
      64.       } else {
      65.         //重用现有的性能记录器
      66.         perfLogger = SessionState.getPerfLogger();
      67.         //由于我们正在重用已编译的计划，因此需要更新其当前运行的开始时间
      68.         plan.setQueryStartTime(perfLogger.getStartTime(PerfLogger.DRIVER_RUN));
      69.       }
      70.       //我们在这里为cxt设置txn管理器的原因是，每个查询都有自己的ctx对象。
      71.       //txn-mgr在同一个Driver实例中共享，该实例可以运行多个查询。
      72.       ctx.setHiveTxnManager(queryTxnMgr);
      73.  
      74.       checkInterrupted("at acquiring the lock.", null, null);
      75.  
      76.       lockAndRespond();
      77.      
      78.       //......省略......
      79.  
      80.       try {
      81.         //执行hql 我们后面专门用一篇博客来研究
      82.         execute();
      83.       } catch (CommandProcessorResponse cpr) {
      84.         rollback(cpr);
      85.         throw cpr;
      86.       }
      87.  
      88.       //如果needRequireLock为false，则此处的发布将不执行任何操作，因为没有锁
      89.       try {
      90.         //由于set autocommit启动了一个隐式txn，请关闭它        if(queryTxnMgr.isImplicitTransactionOpen() || plan.getOperation() == HiveOperation.COMMIT) {
      91.           releaseLocksAndCommitOrRollback(true);
      92.         }
      93.         else if(plan.getOperation() == HiveOperation.ROLLBACK) {
      94.           releaseLocksAndCommitOrRollback(false);
      95.         }
      96.         else {
      97.           //txn（如果有一个已启动）未完成
      98.         }
      99.       } catch (LockException e) {
      100.         throw handleHiveException(e, 12);
      101.       }
      102.  
      103.       perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.DRIVER_RUN);
      104.       queryDisplay.setPerfLogStarts(QueryDisplay.Phase.EXECUTION, perfLogger.getStartTimes());
      105.       queryDisplay.setPerfLogEnds(QueryDisplay.Phase.EXECUTION, perfLogger.getEndTimes());
      106.  
      107.       //获取所有驱动程序运行的钩子并执行它们。
      108.       try {
      109.         hookRunner.runPostDriverHooks(hookContext);
      110.       } catch (Exception e) {
      111.  
      112.       }
      113.       isFinishedWithError = false;
      114.     } finally {
      115.       if (lDrvState.isAborted()) {
      116.         closeInProcess(true);
      117.       } else {
      118.         //正常只释放相关资源ctx、driverContext
      119.         releaseResources();
      120.       }
      121.  
      122.       lDrvState.stateLock.lock();
      123.       try {
      124.         lDrvState.driverState = isFinishedWithError ? DriverState.ERROR : DriverState.EXECUTED;
      125.       } finally {
      126.         lDrvState.stateLock.unlock();
      127.       }
      128.     }
      129.   }
      五、总结
      1、用户在hive客户端输入hql
      2、进行中断操作，终止正在运行的mr作业
      3、解析用户在hive客户端输入的hql（将hql按照字符一个一个处理，遇到 ";" 就会将前面的处理成一个hql 放入列表中）
      4、循环执行hql列表中的每一条hql
      5、从sql语句中剥离注释，并去掉头尾空白符 并按照 '\\s+' 分割成hql数组
      6、判断hql 是 正常的sql（只分析这个） 还是 source 、quit 、 exit 还是 !
      7、获取执行hql的驱动程序（对hql数组的第一个字符串进行转大写操作并匹配对应的驱动程序，默认会返回ReExecDriver）
      8、编译hql
      9、解析hql
      10、语义分析和计划生成
      11、校验计划
      12、获取输出模式并制作查询计划，并设置mapreduce工作流引擎参数
      13、授权检查
              13.1、连接hive的元数据
              13.2、设置输入输出
              13.3、获取表和列的映射
              13.4、添加正在使用的永久UDF
              13.5、通过表扫描运算符检查列授权
              13.6、表授权检查
      14、设置执行计划并执行
    43. 

                    相关阅读:
      
                    
GPTCache使用                            
      
机器人过程自动化(RPA)入门 8. 异常处理、调试和日志记录                            
      
[网鼎杯 2018]Comment                            
      
【Linux常用命令12】搜索命令及特殊字符的使用                            
      
【C++】继承 ⑤ ( public 公有继承 - 示例分析 | protected 保护继承 - 示例分析 | private 私有继承 - 示例分析 )                            
      
java计算机毕业设计即时高校信息发布系统MyBatis+系统+LW文档+源码+调试部署                            
      
内网穿透——搭建私人影音媒体平台                            
      
java175-method类反射机制                            
      
XML序列化与反序列化接口对接实战，看这篇就够了                            
      
sql高级进阶                            
      
                          
                
    44. 
                    原文地址：https://blog.csdn.net/lu070828/article/details/136259577
                
      45.