diff --git a/beeline/src/java/org/apache/hive/beeline/BeeLine.java b/beeline/src/java/org/apache/hive/beeline/BeeLine.java index 45a7e87..c5ca772 100644 --- a/beeline/src/java/org/apache/hive/beeline/BeeLine.java +++ b/beeline/src/java/org/apache/hive/beeline/BeeLine.java @@ -87,6 +87,7 @@ import org.apache.commons.cli.OptionBuilder; import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; +import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.io.IOUtils; import org.apache.hive.beeline.cli.CliOptionsProcessor; import org.apache.hive.jdbc.Utils; @@ -901,27 +902,6 @@ private int executeFile(String fileName, boolean isSourceCMD) { } } - private boolean isSourceCMD(String cmd) { - if (cmd == null || cmd.isEmpty()) - return false; - String[] tokens = tokenizeCmd(cmd); - return tokens[0].equalsIgnoreCase("!source"); - } - - private boolean sourceFile(String cmd) { - String[] tokens = tokenizeCmd(cmd); - String cmd_1 = getFirstCmd(cmd, tokens[0].length()); - File sourceFile = new File(cmd_1); - if (!sourceFile.isFile()) { - return false; - } else { - boolean ret = (executeFile(cmd_1, true) == ERRNO_OK); - // For source command, we should not exit even when meeting some empty line. - setExit(false); - return ret; - } - } - private int execute(ConsoleReader reader, boolean exitOnError) { String line; while (!exit) { @@ -933,10 +913,6 @@ private int execute(ConsoleReader reader, boolean exitOnError) { // trim line line = (line == null) ? null : line.trim(); - if (!isBeeLine) { - line = cliToBeelineCmd(line); - } - if (!dispatch(line) && exitOnError) { return ERRNO_OTHER; } @@ -1043,30 +1019,6 @@ void usage() { } /** - * Extract and clean up the first command in the input. - */ - private String getFirstCmd(String cmd, int length) { - return cmd.substring(length).trim(); - } - - private String cliToBeelineCmd(String cmd) { - if (cmd == null) - return null; - String[] tokens = tokenizeCmd(cmd); - if (cmd.equalsIgnoreCase("quit") || cmd.equalsIgnoreCase("exit")) { - return null; - } else if (tokens[0].equalsIgnoreCase("source")) { - return COMMAND_PREFIX + cmd; - } else if (cmd.startsWith("!")) { - String shell_cmd = cmd.substring(1); - return "!sh " + shell_cmd; - } else { // local mode - // command like dfs - return cmd; - } - } - - /** * Dispatch the specified line to the appropriate {@link CommandHandler}. * * @param line @@ -1088,10 +1040,6 @@ boolean dispatch(String line) { return true; } - if(isSourceCMD(line)){ - return sourceFile(line); - } - line = line.trim(); // save it to the current script, if any @@ -2033,6 +1981,10 @@ void setCompletions() throws SQLException, IOException { } } + public HiveConf getHiveConf(){ + return commands.getHiveConf(); + } + public BeeLineOpts getOpts() { return opts; } @@ -2120,4 +2072,12 @@ void setBatch(List batch) { protected Reflector getReflector() { return reflector; } + + public boolean isBeeLine() { + return isBeeLine; + } + + public void setBeeLine(boolean isBeeLine) { + this.isBeeLine = isBeeLine; + } } diff --git a/beeline/src/java/org/apache/hive/beeline/BeelineVariableSubstitution.java b/beeline/src/java/org/apache/hive/beeline/BeelineVariableSubstitution.java new file mode 100644 index 0000000..ac6fc44 --- /dev/null +++ b/beeline/src/java/org/apache/hive/beeline/BeelineVariableSubstitution.java @@ -0,0 +1,62 @@ +/** + * 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.hive.beeline; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.SystemVariables; + +import java.util.Map; + +public class BeelineVariableSubstitution extends SystemVariables { + private static final Log l4j = LogFactory.getLog(BeelineVariableSubstitution.class); + + private BeeLine beeLine; + + public BeelineVariableSubstitution(BeeLine beeLine) { + this.beeLine = beeLine; + } + + @Override protected String getSubstitute(Configuration conf, String var) { + String val = super.getSubstitute(conf, var); + if (val == null) { + Map vars = beeLine.getCommands().getHiveVariables(); + if (var.startsWith(HIVEVAR_PREFIX)) { + val = vars.get(var.substring(HIVEVAR_PREFIX.length())); + } else { + val = vars.get(var); + } + } + return val; + } + + public String substitute(HiveConf conf, String expr) { + if (expr == null) { + return expr; + } + if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVEVARIABLESUBSTITUTE)) { + l4j.debug("Substitution is on: " + expr); + } else { + return expr; + } + int depth = HiveConf.getIntVar(conf, HiveConf.ConfVars.HIVEVARIABLESUBSTITUTEDEPTH); + return substitute(conf, expr, depth); + } +} diff --git a/beeline/src/java/org/apache/hive/beeline/Commands.java b/beeline/src/java/org/apache/hive/beeline/Commands.java index a42baa3..b882f68 100644 --- a/beeline/src/java/org/apache/hive/beeline/Commands.java +++ b/beeline/src/java/org/apache/hive/beeline/Commands.java @@ -22,6 +22,8 @@ */ package org.apache.hive.beeline; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.SystemVariables; import org.apache.hadoop.io.IOUtils; import java.io.BufferedReader; @@ -32,7 +34,6 @@ import java.io.InputStream; import java.io.InputStreamReader; import java.lang.reflect.Method; -import java.net.MalformedURLException; import java.net.URL; import java.net.URLClassLoader; import java.sql.CallableStatement; @@ -44,9 +45,11 @@ import java.sql.Statement; import java.sql.SQLWarning; import java.util.Arrays; +import java.util.HashMap; import java.util.Iterator; import java.util.LinkedList; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.Set; import java.util.TreeSet; @@ -719,9 +722,7 @@ public boolean sh(String line) { } line = line.substring("sh".length()).trim(); - - // Support variable substitution. HIVE-6791. - // line = new VariableSubstitution().substitute(new HiveConf(BeeLine.class), line.trim()); + line = new BeelineVariableSubstitution(beeLine).substitute(getHiveConf(), line.trim()); try { ShellCmdExecutor executor = new ShellCmdExecutor(line, beeLine.getOutputStream(), @@ -743,6 +744,271 @@ public boolean call(String line) { return execute(line, true); } + public Map getHiveVariables() { + Map result = new HashMap<>(); + Statement stmnt = null; + boolean hasResults; + try { + stmnt = beeLine.createStatement(); + hasResults = stmnt.execute("set"); + if (hasResults) { + ResultSet rs = stmnt.getResultSet(); + BufferedRows rows = new BufferedRows(beeLine, rs); + + while (rows.hasNext()) { + Rows.Row row = (Rows.Row) rows.next(); + if (!row.isMeta) { + result.put(row.values[0], row.values[1]); + } + } + } + } catch (SQLException e) { + beeLine.error(e); + } + return result; + } + + public HiveConf getHiveConf() { + HiveConf conf = new HiveConf(); + Statement stmnt; + boolean hasResults; + try { + stmnt = beeLine.createStatement(); + hasResults = stmnt.execute("set"); + if (hasResults) { + ResultSet rs = stmnt.getResultSet(); + BufferedRows rows = new BufferedRows(beeLine, rs); + while (rows.hasNext()) { + addConf((Rows.Row) rows.next(), conf); + } + } + } catch (SQLException e) { + beeLine.error(e); + } + return conf; + } + + private void addConf(Rows.Row r, HiveConf hiveConf) { + if (r.isMeta) { + return; + } + if (r.values == null || r.values[0] == null || r.values[0].isEmpty()) { + return; + } + String val = r.values[0]; + if (r.values[0].startsWith(SystemVariables.SYSTEM_PREFIX) || r.values[0].startsWith(SystemVariables.ENV_PREFIX)) { + return; + } else { + String[] kv = val.split("=",2); + hiveConf.set(kv[0], kv[1]); + } + } + + /** + * Extract and clean up the first command in the input. + */ + private String getFirstCmd(String cmd, int length) { + return cmd.substring(length).trim(); + } + + private String[] tokenizeCmd(String cmd) { + return cmd.split("\\s+"); + } + + private boolean isSourceCMD(String cmd) { + if (cmd == null || cmd.isEmpty()) + return false; + String[] tokens = tokenizeCmd(cmd); + return tokens[0].equalsIgnoreCase("!source"); + } + + private boolean sourceFile(String cmd) { + String[] tokens = tokenizeCmd(cmd); + String cmd_1 = getFirstCmd(cmd, tokens[0].length()); + cmd_1 = new BeelineVariableSubstitution(beeLine).substitute(getHiveConf(), cmd_1); + File sourceFile = new File(cmd_1); + if (!sourceFile.isFile()) { + return false; + } else { + boolean ret; + try { + ret = sourceFileInternal(sourceFile, true); + } catch (IOException e) { + beeLine.error(e); + return false; + } + return ret; + } + } + + private boolean sourceFileInternal(File sourceFile, boolean call) throws IOException { + BufferedReader reader = new BufferedReader(new FileReader(sourceFile)); + String extra = reader.readLine(); + String lines = null; + while (extra != null) { + if (beeLine.isComment(extra)) { + continue; + } + if (lines == null) { + lines = extra; + } else { + lines += "\n" + extra; + } + extra = reader.readLine(); + } + String[] cmds = lines.split(";"); + for (String c : cmds) { + if (!executeInternal(c, false)) { + return false; + } + } + return true; + } + + private String cliToBeelineCmd(String cmd) { + if (cmd == null) + return null; + String[] tokens = tokenizeCmd(cmd); + if (tokens[0].equalsIgnoreCase("source")) { + return BeeLine.COMMAND_PREFIX + cmd; + } else if (cmd.startsWith("!")) { + String shell_cmd = cmd.substring(1); + return "!sh " + shell_cmd; + } else { // local mode + // command like dfs + return cmd; + } + } + + // Return false only occurred error when execution the sql and the sql should follow the rules + // of beeline. + private boolean executeInternal(String sql, boolean call){ + if (sql == null || sql.length() == 0) { + return true; + } + + if (beeLine.isComment(sql)) { + //skip this and rest cmds in the line + return true; + } + + if (sql.startsWith(BeeLine.COMMAND_PREFIX)) { + sql = sql.substring(1); + } + + String prefix = call ? "call" : "sql"; + + if (sql.startsWith(prefix)) { + sql = sql.substring(prefix.length()); + } + + // batch statements? + if (beeLine.getBatch() != null) { + beeLine.getBatch().add(sql); + return true; + } + + try { + Statement stmnt = null; + boolean hasResults; + Thread logThread = null; + + try { + long start = System.currentTimeMillis(); + + if (call) { + stmnt = beeLine.getDatabaseConnection().getConnection().prepareCall(sql); + hasResults = ((CallableStatement) stmnt).execute(); + } else { + stmnt = beeLine.createStatement(); + if (beeLine.getOpts().isSilent()) { + hasResults = stmnt.execute(sql); + } else { + logThread = new Thread(createLogRunnable(stmnt)); + logThread.setDaemon(true); + logThread.start(); + hasResults = stmnt.execute(sql); + logThread.interrupt(); + logThread.join(DEFAULT_QUERY_PROGRESS_THREAD_TIMEOUT); + } + } + + beeLine.showWarnings(); + + if (hasResults) { + do { + ResultSet rs = stmnt.getResultSet(); + try { + int count = beeLine.print(rs); + long end = System.currentTimeMillis(); + + beeLine.info( + beeLine.loc("rows-selected", count) + " " + beeLine.locElapsedTime(end - start)); + } finally { + if (logThread != null) { + logThread.join(DEFAULT_QUERY_PROGRESS_THREAD_TIMEOUT); + showRemainingLogsIfAny(stmnt); + logThread = null; + } + rs.close(); + } + } while (BeeLine.getMoreResults(stmnt)); + } else { + int count = stmnt.getUpdateCount(); + long end = System.currentTimeMillis(); + beeLine.info( + beeLine.loc("rows-affected", count) + " " + beeLine.locElapsedTime(end - start)); + } + } finally { + if (logThread != null) { + if (!logThread.isInterrupted()) { + logThread.interrupt(); + } + logThread.join(DEFAULT_QUERY_PROGRESS_THREAD_TIMEOUT); + showRemainingLogsIfAny(stmnt); + } + if (stmnt != null) { + stmnt.close(); + } + } + } catch (Exception e) { + return beeLine.error(e); + } + beeLine.showWarnings(); + return true; + } + + public String handleMultiLineCmd(String line) throws IOException { + //When using -e, console reader is not initialized and command is a single line + while (beeLine.getConsoleReader() != null && !(line.trim().endsWith(";")) && beeLine.getOpts() + .isAllowMultiLineCommand()) { + + if (!beeLine.getOpts().isSilent()) { + StringBuilder prompt = new StringBuilder(beeLine.getPrompt()); + for (int i = 0; i < prompt.length() - 1; i++) { + if (prompt.charAt(i) != '>') { + prompt.setCharAt(i, i % 2 == 0 ? '.' : ' '); + } + } + } + + String extra; + if (beeLine.getOpts().isSilent() && beeLine.getOpts().getScriptFile() != null) { + extra = beeLine.getConsoleReader().readLine(null, jline.console.ConsoleReader.NULL_MASK); + } else { + extra = beeLine.getConsoleReader().readLine(beeLine.getPrompt()); + } + + if (extra == null) { //it happens when using -f and the line of cmds does not end with ; + break; + } + if (!beeLine.isComment(extra)) { + line += "\n" + extra; + } + } + return line; + } + private boolean execute(String line, boolean call) { if (line == null || line.length() == 0) { return false; // ??? @@ -756,33 +1022,7 @@ private boolean execute(String line, boolean call) { // use multiple lines for statements not terminated by ";" try { - //When using -e, console reader is not initialized and command is a single line - while (beeLine.getConsoleReader() != null && !(line.trim().endsWith(";")) - && beeLine.getOpts().isAllowMultiLineCommand()) { - - if (!beeLine.getOpts().isSilent()) { - StringBuilder prompt = new StringBuilder(beeLine.getPrompt()); - for (int i = 0; i < prompt.length() - 1; i++) { - if (prompt.charAt(i) != '>') { - prompt.setCharAt(i, i % 2 == 0 ? '.' : ' '); - } - } - } - - String extra = null; - if (beeLine.getOpts().isSilent() && beeLine.getOpts().getScriptFile() != null) { - extra = beeLine.getConsoleReader().readLine(null, jline.console.ConsoleReader.NULL_MASK); - } else { - extra = beeLine.getConsoleReader().readLine(beeLine.getPrompt()); - } - - if (extra == null) { //it happens when using -f and the line of cmds does not end with ; - break; - } - if (!beeLine.isComment(extra)) { - line += "\n" + extra; - } - } + line = handleMultiLineCmd(line); } catch (Exception e) { beeLine.handleException(e); } @@ -796,93 +1036,23 @@ private boolean execute(String line, boolean call) { for (int i = 0; i < cmds.length; i++) { String sql = cmds[i].trim(); if (sql.length() != 0) { - if (beeLine.isComment(sql)) { - //skip this and rest cmds in the line - break; - } - if (sql.startsWith(BeeLine.COMMAND_PREFIX)) { - sql = sql.substring(1); - } - - String prefix = call ? "call" : "sql"; - - if (sql.startsWith(prefix)) { - sql = sql.substring(prefix.length()); + if (!beeLine.isBeeLine()) { + sql = cliToBeelineCmd(sql); + if (sql.equalsIgnoreCase("quit") || sql.equalsIgnoreCase("exit")) { + beeLine.setExit(true); + return true; + } } - // batch statements? - if (beeLine.getBatch() != null) { - beeLine.getBatch().add(sql); + // is source CMD + if (isSourceCMD(sql)) { + sourceFile(sql); continue; } - try { - Statement stmnt = null; - boolean hasResults; - Thread logThread = null; - - try { - long start = System.currentTimeMillis(); - - if (call) { - stmnt = beeLine.getDatabaseConnection().getConnection().prepareCall(sql); - hasResults = ((CallableStatement) stmnt).execute(); - } else { - stmnt = beeLine.createStatement(); - if (beeLine.getOpts().isSilent()) { - hasResults = stmnt.execute(sql); - } else { - logThread = new Thread(createLogRunnable(stmnt)); - logThread.setDaemon(true); - logThread.start(); - hasResults = stmnt.execute(sql); - logThread.interrupt(); - logThread.join(DEFAULT_QUERY_PROGRESS_THREAD_TIMEOUT); - } - } - - beeLine.showWarnings(); - - if (hasResults) { - do { - ResultSet rs = stmnt.getResultSet(); - try { - int count = beeLine.print(rs); - long end = System.currentTimeMillis(); - - beeLine.info(beeLine.loc("rows-selected", count) + " " - + beeLine.locElapsedTime(end - start)); - } finally { - if (logThread != null) { - logThread.join(DEFAULT_QUERY_PROGRESS_THREAD_TIMEOUT); - showRemainingLogsIfAny(stmnt); - logThread = null; - } - rs.close(); - } - } while (BeeLine.getMoreResults(stmnt)); - } else { - int count = stmnt.getUpdateCount(); - long end = System.currentTimeMillis(); - beeLine.info(beeLine.loc("rows-affected", count) - + " " + beeLine.locElapsedTime(end - start)); - } - } finally { - if (logThread != null) { - if (!logThread.isInterrupted()) { - logThread.interrupt(); - } - logThread.join(DEFAULT_QUERY_PROGRESS_THREAD_TIMEOUT); - showRemainingLogsIfAny(stmnt); - } - if (stmnt != null) { - stmnt.close(); - } - } - } catch (Exception e) { - return beeLine.error(e); + if (!executeInternal(sql, call)) { + return false; } - beeLine.showWarnings(); } } return true; diff --git a/beeline/src/test/org/apache/hive/beeline/cli/TestHiveCli.java b/beeline/src/test/org/apache/hive/beeline/cli/TestHiveCli.java index 6cbb030..8764f7e 100644 --- a/beeline/src/test/org/apache/hive/beeline/cli/TestHiveCli.java +++ b/beeline/src/test/org/apache/hive/beeline/cli/TestHiveCli.java @@ -42,6 +42,10 @@ private final static String SOURCE_CONTEXT = "create table if not exists test.testSrcTbl(a string, b string);"; + private final static String SOURCE_CONTEXT2 = + "create table if not exists test.testSrcTbl2(a string);"; + private final static String SOURCE_CONTEXT3 = + "create table if not exists test.testSrcTbl3(a string);"; final static String CMD = "create database if not exists test;\ncreate table if not exists test.testTbl(a string, b " + "string);\n"; @@ -60,7 +64,7 @@ private void executeCMD(String[] args, String input, int retCode) { inputStream = IOUtils.toInputStream(input); } ret = cli.runWithArgs(args, inputStream); - } catch (IOException e) { + } catch (Throwable e) { LOG.error("Failed to execute command due to the error: " + e); } finally { if (retCode != ret) { @@ -78,7 +82,7 @@ private void verifyCMD(String CMD, String keywords, OutputStream os, String[] op } @Test public void testInValidCmd() { - verifyCMD("!lss\n", "Failed to execute lss", errS, null, ERRNO_OK); + verifyCMD("!lss\n", "Unknown command: lss", errS, null, ERRNO_OK); } @Test public void testHelp() { @@ -86,7 +90,7 @@ private void verifyCMD(String CMD, String keywords, OutputStream os, String[] op } @Test public void testInvalidDatabaseOptions() { - verifyCMD("\nshow tables\nquit\n", "Database does not exist: invalidDB", errS, + verifyCMD("\nshow tables;\nquit;\n", "Database does not exist: invalidDB", errS, new String[] { "--database", "invalidDB" }, ERRNO_OK); } @@ -97,8 +101,16 @@ private void verifyCMD(String CMD, String keywords, OutputStream os, String[] op @Test public void testSourceCmd() { File f = generateTmpFile(SOURCE_CONTEXT); - verifyCMD("source " + f.getPath() + "\n" + "desc testSrcTbl\n" + "quit\n", "col_name", os, + verifyCMD("source " + f.getPath() + ";" + "desc testSrcTbl;\nquit;\n", "col_name", os, new String[] { "--database", "test" }, ERRNO_OK); + f.delete(); + } + + @Test public void testSourceCmd2() { + File f = generateTmpFile(SOURCE_CONTEXT3); + verifyCMD("source " + f.getPath() + ";" + "desc testSrcTbl3;\nquit;\n", "col_name", os, + new String[] { "--database", "test" }, ERRNO_OK); + f.delete(); } @Test public void testSqlFromCmd() { @@ -119,6 +131,19 @@ private void verifyCMD(String CMD, String keywords, OutputStream os, String[] op verifyCMD(null, "Unrecognized option: -k", errS, new String[] { "-k" }, ERRNO_ARGS); } + @Test public void testVariables() { + verifyCMD("set system:xxx=5;\nset system:yyy=${system:xxx};\nset system:yyy;", "", os, null, + ERRNO_OK); + } + + @Test public void testVariablesForSource() { + File f = generateTmpFile(SOURCE_CONTEXT2); + verifyCMD( + "set hiveconf:zzz=" + f.getAbsolutePath() + ";\nsource ${hiveconf:zzz};\ndesc testSrcTbl2;", + "col_name", os, new String[] { "--database", "test" }, ERRNO_OK); + f.delete(); + } + private void redirectOutputStream() { // Setup output stream to redirect output to os = new ByteArrayOutputStream();