diff --git beeline/src/java/org/apache/hive/beeline/Commands.java beeline/src/java/org/apache/hive/beeline/Commands.java index 08d53ca..407e018 100644 --- beeline/src/java/org/apache/hive/beeline/Commands.java +++ beeline/src/java/org/apache/hive/beeline/Commands.java @@ -1003,6 +1003,15 @@ private boolean executeInternal(String sql, boolean call) { beeLine.showWarnings(); if (hasResults) { + OutputFile outputFile = beeLine.getRecordOutputFile(); + if (beeLine.isTestMode() && outputFile != null && outputFile.isActiveConverter()) { + outputFile.fetchStarted(); + if (!sql.trim().toLowerCase().startsWith("explain")) { + outputFile.foundQuery(true); + } else { + outputFile.foundQuery(false); + } + } do { ResultSet rs = stmnt.getResultSet(); try { @@ -1020,6 +1029,9 @@ private boolean executeInternal(String sql, boolean call) { rs.close(); } } while (BeeLine.getMoreResults(stmnt)); + if (beeLine.isTestMode() && outputFile != null && outputFile.isActiveConverter()) { + outputFile.fetchFinished(); + } } else { int count = stmnt.getUpdateCount(); long end = System.currentTimeMillis(); diff --git beeline/src/java/org/apache/hive/beeline/OutputFile.java beeline/src/java/org/apache/hive/beeline/OutputFile.java index 1014af3..fa957fa 100644 --- beeline/src/java/org/apache/hive/beeline/OutputFile.java +++ beeline/src/java/org/apache/hive/beeline/OutputFile.java @@ -23,17 +23,87 @@ package org.apache.hive.beeline; import java.io.File; -import java.io.FileWriter; import java.io.IOException; -import java.io.PrintWriter; +import java.io.PrintStream; + +import org.apache.hadoop.hive.common.io.DigestPrintStream; +import org.apache.hadoop.hive.common.io.FetchConverter; +import org.apache.hadoop.hive.common.io.SortAndDigestPrintStream; +import org.apache.hadoop.hive.common.io.SortPrintStream; public class OutputFile { - final File file; - final PrintWriter out; + private final File file; + private PrintStream out; + private boolean isActiveFetchConverter = false; public OutputFile(String filename) throws IOException { file = new File(filename); - out = new PrintWriter(new FileWriter(file)); + out = new PrintStream(file, "UTF-8"); + } + + /** + * Set converter for the output. Used only for testing. + * @param converter The type of the converter to use + * @throws Exception In case of an error in stream creation + */ + void setConverter(SupportedConverter converter) throws Exception { + switch (converter) { + case SORT_QUERY_RESULTS: + out = new SortPrintStream(out, "UTF-8"); + break; + case HASH_QUERY_RESULTS: + out = new DigestPrintStream(out, "UTF-8"); + break; + case SORT_AND_HASH_QUERY_RESULTS: + out = new SortAndDigestPrintStream(out, "UTF-8"); + break; + default: + // No wrapping is needed + } + isActiveFetchConverter = (out instanceof FetchConverter); + } + + /** + * Returns true if a FetchConverter is defined for writing the results. Should be used only for + * testing. + * @return True if a FetchConverter is active + */ + boolean isActiveConverter() { + return isActiveFetchConverter; + } + + /** + * Indicates that result fetching is started, and the converter should be activated. The + * Converter starts to collect the data when the fetch is started, and prints out the + * converted data when the fetch is finished. Converter will collect data only if + * fetchStarted, and foundQuery is true. + */ + void fetchStarted() { + if (isActiveFetchConverter) { + ((FetchConverter) out).fetchStarted(); + } + } + + /** + * Indicates that the following data will be a query result, and the converter should be + * activated. Converter will collect the data only if fetchStarted, and foundQuery is true. + * @param foundQuery The following data will be a query result (true) or not (false) + */ + void foundQuery(boolean foundQuery) { + if (isActiveFetchConverter) { + ((FetchConverter) out).foundQuery(foundQuery); + } + } + + /** + * Indicates that the previously collected data should be converted and written. Converter + * starts to collect the data when the fetch is started, and prints out the converted data when + * the fetch is finished. + */ + void fetchFinished() { + if (isActiveFetchConverter) { + ((FetchConverter) out).fetchFinished(); + } } @Override @@ -56,4 +126,14 @@ public void print(String command) { public void close() throws IOException { out.close(); } + + /** + * The supported type of converters. All of the points to a specific FetchConverter class. + */ + public enum SupportedConverter { + SORT_QUERY_RESULTS, + HASH_QUERY_RESULTS, + SORT_AND_HASH_QUERY_RESULTS, + NONE + } } diff --git itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java index 8c7057c..f95c3ca 100644 --- itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java +++ itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java @@ -23,6 +23,7 @@ import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.QTestProcessExecResult; import org.apache.hadoop.hive.ql.hooks.PreExecutePrinter; +import org.apache.hive.beeline.OutputFile.SupportedConverter; import org.apache.hive.beeline.QFile; import org.apache.hive.beeline.QFile.QFileBuilder; import org.apache.hive.beeline.QFileBeeLineClient; @@ -118,7 +119,8 @@ protected void runInfraScript(File script, File beeLineOutput, File log) "set test.script.dir=" + testScriptDirectory + ";", "!run " + script, }, - log); + log, + SupportedConverter.NONE); } catch (Exception e) { throw new SQLException("Error running infra script: " + script + "\nCheck the following logs for details:\n - " + beeLineOutput + "\n - " + log, e); diff --git itests/util/src/main/java/org/apache/hive/beeline/QFile.java itests/util/src/main/java/org/apache/hive/beeline/QFile.java index 0bde529..5fe6656 100644 --- itests/util/src/main/java/org/apache/hive/beeline/QFile.java +++ itests/util/src/main/java/org/apache/hive/beeline/QFile.java @@ -23,6 +23,7 @@ import org.apache.hadoop.hive.ql.QTestUtil; import org.apache.hadoop.util.Shell; import org.apache.hive.common.util.StreamPrinter; +import org.apache.hive.beeline.OutputFile.SupportedConverter; import java.io.ByteArrayOutputStream; import java.io.File; @@ -72,6 +73,7 @@ private static RegexFilterSet staticFilterSet = getStaticFilterSet(); private RegexFilterSet specificFilterSet; private boolean rewriteSourceTables; + private SupportedConverter converter; private QFile() {} @@ -107,6 +109,10 @@ public File getAfterExecuteLogFile() { return afterExecuteLogFile; } + public SupportedConverter getConverter() { + return converter; + } + public String getDebugHint() { return String.format(DEBUG_HINT, inputFile, rawOutputFile, outputFile, expectedOutputFile, logFile, beforeExecuteLogFile, afterExecuteLogFile, @@ -327,6 +333,17 @@ public QFile getQFile(String name) throws IOException { .addFilter("(PREHOOK|POSTHOOK): (Output|Input): " + name + "@", "$1: $2: default@") .addFilter("name(:?) " + name + "\\.(.*)\n", "name$1 default.$2\n") .addFilter("/" + name + ".db/", "/"); + result.converter = SupportedConverter.NONE; + String input = FileUtils.readFileToString(result.inputFile, "UTF-8"); + if (input.contains("-- SORT_QUERY_RESULTS")) { + result.converter = SupportedConverter.SORT_QUERY_RESULTS; + } + if (input.contains("-- HASH_QUERY_RESULTS")) { + result.converter = SupportedConverter.HASH_QUERY_RESULTS; + } + if (input.contains("-- SORT_AND_HASH_QUERY_RESULTS")) { + result.converter = SupportedConverter.SORT_AND_HASH_QUERY_RESULTS; + } return result; } } diff --git itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java index f1b53f7..8744783 100644 --- itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java +++ itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java @@ -23,6 +23,8 @@ import java.io.PrintStream; import java.sql.SQLException; +import org.apache.hive.beeline.OutputFile.SupportedConverter; + /** * QFile test client using BeeLine. It can be used to submit a list of command strings, or a QFile. */ @@ -50,11 +52,13 @@ protected QFileBeeLineClient(String jdbcUrl, String jdbcDriver, String username, }); } - public void execute(String[] commands, File resultFile) throws SQLException { + public void execute(String[] commands, File resultFile, SupportedConverter converter) + throws Exception { beeLine.runCommands( new String[] { "!record " + resultFile.getAbsolutePath() }); + beeLine.getRecordOutputFile().setConverter(converter); int lastSuccessfulCommand = beeLine.runCommands(commands); if (commands.length != lastSuccessfulCommand) { @@ -64,7 +68,7 @@ public void execute(String[] commands, File resultFile) throws SQLException { beeLine.runCommands(new String[] {"!record"}); } - private void beforeExecute(QFile qFile) throws SQLException { + private void beforeExecute(QFile qFile) throws Exception { execute( new String[] { "!set outputformat tsv2", @@ -79,11 +83,12 @@ private void beforeExecute(QFile qFile) throws SQLException { "set hive.in.test.short.logs=true;", "set hive.in.test.remove.logs=false;", }, - qFile.getBeforeExecuteLogFile()); + qFile.getBeforeExecuteLogFile(), + SupportedConverter.NONE); beeLine.setIsTestMode(true); } - private void afterExecute(QFile qFile) throws SQLException { + private void afterExecute(QFile qFile) throws Exception { beeLine.setIsTestMode(false); execute( new String[] { @@ -95,13 +100,14 @@ private void afterExecute(QFile qFile) throws SQLException { "USE default;", "DROP DATABASE IF EXISTS `" + qFile.getName() + "` CASCADE;", }, - qFile.getAfterExecuteLogFile()); + qFile.getAfterExecuteLogFile(), + SupportedConverter.NONE); } - public void execute(QFile qFile) throws SQLException, IOException { + public void execute(QFile qFile) throws Exception { beforeExecute(qFile); String[] commands = beeLine.getCommands(qFile.getInputFile()); - execute(qFile.filterCommands(commands), qFile.getRawOutputFile()); + execute(qFile.filterCommands(commands), qFile.getRawOutputFile(), qFile.getConverter()); afterExecute(qFile); } diff --git ql/src/test/results/clientpositive/beeline/smb_mapjoin_1.q.out ql/src/test/results/clientpositive/beeline/smb_mapjoin_1.q.out index c943b03..40df1c3 100644 --- ql/src/test/results/clientpositive/beeline/smb_mapjoin_1.q.out +++ ql/src/test/results/clientpositive/beeline/smb_mapjoin_1.q.out @@ -150,10 +150,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_2 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 NULL NULL 3 val_3 NULL NULL 4 val_4 NULL NULL 5 val_5 NULL NULL -10 val_10 NULL NULL PREHOOK: query: explain select /*+mapjoin(a)*/ * from smb_bucket_1 a right outer join smb_bucket_2 b on a.key = b.key PREHOOK: type: QUERY @@ -259,10 +259,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_2 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 NULL NULL 3 val_3 NULL NULL 4 val_4 NULL NULL 5 val_5 NULL NULL -10 val_10 NULL NULL NULL NULL 20 val_20 NULL NULL 23 val_23 NULL NULL 25 val_25 @@ -371,10 +371,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_2 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 NULL NULL 3 val_3 NULL NULL 4 val_4 NULL NULL 5 val_5 NULL NULL -10 val_10 NULL NULL PREHOOK: query: explain select /*+mapjoin(b)*/ * from smb_bucket_1 a right outer join smb_bucket_2 b on a.key = b.key PREHOOK: type: QUERY @@ -480,10 +480,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_2 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 NULL NULL 3 val_3 NULL NULL 4 val_4 NULL NULL 5 val_5 NULL NULL -10 val_10 NULL NULL NULL NULL 20 val_20 NULL NULL 23 val_23 NULL NULL 25 val_25 diff --git ql/src/test/results/clientpositive/beeline/smb_mapjoin_2.q.out ql/src/test/results/clientpositive/beeline/smb_mapjoin_2.q.out index 1ea6553..7840905 100644 --- ql/src/test/results/clientpositive/beeline/smb_mapjoin_2.q.out +++ ql/src/test/results/clientpositive/beeline/smb_mapjoin_2.q.out @@ -99,8 +99,8 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -4 val_4 4 val_4 10 val_10 10 val_10 +4 val_4 4 val_4 PREHOOK: query: explain select /*+mapjoin(a)*/ * from smb_bucket_1 a left outer join smb_bucket_3 b on a.key = b.key PREHOOK: type: QUERY @@ -152,10 +152,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 10 val_10 3 val_3 NULL NULL 4 val_4 4 val_4 5 val_5 NULL NULL -10 val_10 10 val_10 PREHOOK: query: explain select /*+mapjoin(a)*/ * from smb_bucket_1 a right outer join smb_bucket_3 b on a.key = b.key PREHOOK: type: QUERY @@ -206,8 +206,8 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -4 val_4 4 val_4 10 val_10 10 val_10 +4 val_4 4 val_4 NULL NULL 17 val_17 NULL NULL 19 val_19 NULL NULL 20 val_20 @@ -263,10 +263,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 10 val_10 3 val_3 NULL NULL 4 val_4 4 val_4 5 val_5 NULL NULL -10 val_10 10 val_10 NULL NULL 17 val_17 NULL NULL 19 val_19 NULL NULL 20 val_20 @@ -324,8 +324,8 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -4 val_4 4 val_4 10 val_10 10 val_10 +4 val_4 4 val_4 PREHOOK: query: explain select /*+mapjoin(b)*/ * from smb_bucket_1 a left outer join smb_bucket_3 b on a.key = b.key PREHOOK: type: QUERY @@ -377,10 +377,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 10 val_10 3 val_3 NULL NULL 4 val_4 4 val_4 5 val_5 NULL NULL -10 val_10 10 val_10 PREHOOK: query: explain select /*+mapjoin(b)*/ * from smb_bucket_1 a right outer join smb_bucket_3 b on a.key = b.key PREHOOK: type: QUERY @@ -431,8 +431,8 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -4 val_4 4 val_4 10 val_10 10 val_10 +4 val_4 4 val_4 NULL NULL 17 val_17 NULL NULL 19 val_19 NULL NULL 20 val_20 @@ -488,10 +488,10 @@ POSTHOOK: Input: default@smb_bucket_1 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### 1 val_1 NULL NULL +10 val_10 10 val_10 3 val_3 NULL NULL 4 val_4 4 val_4 5 val_5 NULL NULL -10 val_10 10 val_10 NULL NULL 17 val_17 NULL NULL 19 val_19 NULL NULL 20 val_20 diff --git ql/src/test/results/clientpositive/beeline/smb_mapjoin_3.q.out ql/src/test/results/clientpositive/beeline/smb_mapjoin_3.q.out index f639ba4..cda600b 100644 --- ql/src/test/results/clientpositive/beeline/smb_mapjoin_3.q.out +++ ql/src/test/results/clientpositive/beeline/smb_mapjoin_3.q.out @@ -205,12 +205,12 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_2 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -NULL NULL 4 val_4 +20 val_20 20 val_20 +23 val_23 23 val_23 NULL NULL 10 val_10 NULL NULL 17 val_17 NULL NULL 19 val_19 -20 val_20 20 val_20 -23 val_23 23 val_23 +NULL NULL 4 val_4 PREHOOK: query: explain select /*+mapjoin(a)*/ * from smb_bucket_2 a full outer join smb_bucket_3 b on a.key = b.key PREHOOK: type: QUERY @@ -261,14 +261,14 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_2 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -NULL NULL 4 val_4 -NULL NULL 10 val_10 -NULL NULL 17 val_17 -NULL NULL 19 val_19 20 val_20 20 val_20 23 val_23 23 val_23 25 val_25 NULL NULL 30 val_30 NULL NULL +NULL NULL 10 val_10 +NULL NULL 17 val_17 +NULL NULL 19 val_19 +NULL NULL 4 val_4 PREHOOK: query: explain select /*+mapjoin(b)*/ * from smb_bucket_2 a join smb_bucket_3 b on a.key = b.key PREHOOK: type: QUERY @@ -428,12 +428,12 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_2 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -NULL NULL 4 val_4 +20 val_20 20 val_20 +23 val_23 23 val_23 NULL NULL 10 val_10 NULL NULL 17 val_17 NULL NULL 19 val_19 -20 val_20 20 val_20 -23 val_23 23 val_23 +NULL NULL 4 val_4 PREHOOK: query: explain select /*+mapjoin(b)*/ * from smb_bucket_2 a full outer join smb_bucket_3 b on a.key = b.key PREHOOK: type: QUERY @@ -484,11 +484,11 @@ POSTHOOK: type: QUERY POSTHOOK: Input: default@smb_bucket_2 POSTHOOK: Input: default@smb_bucket_3 #### A masked pattern was here #### -NULL NULL 4 val_4 -NULL NULL 10 val_10 -NULL NULL 17 val_17 -NULL NULL 19 val_19 20 val_20 20 val_20 23 val_23 23 val_23 25 val_25 NULL NULL 30 val_30 NULL NULL +NULL NULL 10 val_10 +NULL NULL 17 val_17 +NULL NULL 19 val_19 +NULL NULL 4 val_4