diff --git a/common/src/java/org/apache/hadoop/hive/common/io/CachingPrintStream.java b/common/src/java/org/apache/hadoop/hive/common/io/CachingPrintStream.java index 70dab87dffd5..40b0e1a6ffbc 100644 --- a/common/src/java/org/apache/hadoop/hive/common/io/CachingPrintStream.java +++ b/common/src/java/org/apache/hadoop/hive/common/io/CachingPrintStream.java @@ -20,6 +20,7 @@ import java.io.FileNotFoundException; import java.io.OutputStream; +import java.io.PrintStream; import java.io.UnsupportedEncodingException; import java.util.ArrayList; import java.util.List; @@ -43,7 +44,12 @@ public CachingPrintStream(OutputStream out) { @Override public void println(String out) { output.add(out); - super.println(out); + // Delegate to a wrapped PrintStream so downstream transforms (e.g. QTestFetchConverter) apply. + if (this.out instanceof PrintStream ps && ps != this) { + ps.println(out); + } else { + super.println(out); + } } @Override diff --git a/common/src/java/org/apache/hadoop/hive/common/io/FetchConverter.java b/common/src/java/org/apache/hadoop/hive/common/io/FetchConverter.java index 73298d9bdb22..26b7c0963dea 100644 --- a/common/src/java/org/apache/hadoop/hive/common/io/FetchConverter.java +++ b/common/src/java/org/apache/hadoop/hive/common/io/FetchConverter.java @@ -19,6 +19,7 @@ package org.apache.hadoop.hive.common.io; import java.io.OutputStream; +import java.io.PrintStream; import java.io.UnsupportedEncodingException; public abstract class FetchConverter extends SessionStream implements FetchCallback { @@ -48,8 +49,14 @@ public void println(String out) { } } - protected final void printDirect(String out) { - super.println(out); + protected final void printDirect(String line) { + // Delegate to a wrapped PrintStream so downstream transforms (e.g. QTestFetchConverter) apply + // when this converter is stacked above them (sort/hash then mask). + if (this.out instanceof PrintStream ps && ps != this) { + ps.println(line); + } else { + super.println(line); + } } protected final boolean byPass() { diff --git a/common/src/java/org/apache/hadoop/hive/common/io/QTestFetchConverter.java b/common/src/java/org/apache/hadoop/hive/common/io/QTestFetchConverter.java index 59fc47060fe7..b93bc7d85fc3 100644 --- a/common/src/java/org/apache/hadoop/hive/common/io/QTestFetchConverter.java +++ b/common/src/java/org/apache/hadoop/hive/common/io/QTestFetchConverter.java @@ -43,7 +43,34 @@ public QTestFetchConverter(OutputStream out, boolean autoFlush, String encoding, @Override public void println(String str) { - inner.println(transformation.apply(str)); + // PREHOOK/POSTHOOK may send multiple lines separated by \n in one println, mask each + // separately. Query fetch rows must stay intact so SORT_QUERY_RESULTS keeps row order. + if (str.indexOf('\n') >= 0 && shouldSplitMultiline()) { + int start = 0; + for (int i = 0; i < str.length(); i++) { + if (str.charAt(i) == '\n') { + writeTransformedLine(str.substring(start, i)); + start = i + 1; + } + } + writeTransformedLine(str.substring(start)); + return; + } + writeTransformedLine(str); + } + + private boolean shouldSplitMultiline() { + if (inner instanceof FetchConverter fetchConverter) { + return !(fetchConverter.queryfound && fetchConverter.fetchStarted); + } + return true; + } + + private void writeTransformedLine(String line) { + String transformed = transformation.apply(line); + if (transformed != null) { + inner.println(transformed); + } } @Override diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java b/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java index 9bdd63d675a6..62f306382798 100644 --- a/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java +++ b/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java @@ -21,6 +21,7 @@ import java.io.File; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.sql.SQLException; import java.util.ArrayList; import java.util.Arrays; @@ -35,6 +36,7 @@ import org.apache.commons.io.FileUtils; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.HiveConf.ConfVars; import org.apache.hadoop.hive.conf.HiveConfUtil; import org.apache.hadoop.hive.ql.QTestProcessExecResult; import org.apache.hadoop.hive.ql.QTestUtil; @@ -227,6 +229,11 @@ private void runTest(QFile qFile, List> preCommands) throws Excep long startTime = System.currentTimeMillis(); System.err.println(">>> STARTED " + qFile.getName()); + qOutProcessor.initMasks(FileUtils.readFileToString(qFile.getInputFile(), StandardCharsets.UTF_8)); + setupAdditionalPartialMasks(); + QOutProcessor.MaskingFoldState maskingFoldState = new QOutProcessor.MaskingFoldState(); + beeLineClient.setOutputMasking(qOutProcessor, maskingFoldState); + beeLineClient.execute(qFile, preCommands); long queryEndTime = System.currentTimeMillis(); @@ -239,7 +246,6 @@ private void runTest(QFile qFile, List> preCommands) throws Excep + "ms"); if (!overwrite) { - qOutProcessor.maskPatterns(qFile.getOutputFile().getPath()); QTestProcessExecResult result = qFile.compareResults(); long compareEndTime = System.currentTimeMillis(); @@ -259,13 +265,40 @@ private void runTest(QFile qFile, List> preCommands) throws Excep qFile.overwriteResults(); System.err.println(">>> PASSED " + qFile.getName()); } + resetAdditionalPartialMasks(); } catch (Exception e) { + resetAdditionalPartialMasks(); throw new Exception("Exception running or analyzing the results of the query file: " + qFile + "\n" + qFile.getDebugHint(), e); } } + private void setupAdditionalPartialMasks() { + if (miniHS2 == null) { + return; + } + HiveConf conf = miniHS2.getHiveConf(); + String patternStr = HiveConf.getVar(conf, ConfVars.HIVE_ADDITIONAL_PARTIAL_MASKS_PATTERN); + String replacementStr = HiveConf.getVar(conf, ConfVars.HIVE_ADDITIONAL_PARTIAL_MASKS_REPLACEMENT_TEXT); + if (patternStr != null && replacementStr != null && !replacementStr.isEmpty() + && !patternStr.isEmpty()) { + String[] patterns = patternStr.split(","); + String[] replacements = replacementStr.split(","); + if (patterns.length != replacements.length) { + throw new RuntimeException("Count mismatch for additional partial masks and their replacements"); + } + for (int i = 0; i < patterns.length; i++) { + qOutProcessor.addPatternWithMaskComment(patterns[i], + String.format("### %s ###", replacements[i])); + } + } + } + + private void resetAdditionalPartialMasks() { + qOutProcessor.resetPatternwithMaskComments(); + } + public void runTest(QFile qFile) throws Exception { runTest(qFile, null); } diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreCliDriver.java b/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreCliDriver.java index 8f4e9ad1a626..31c488cf0eab 100644 --- a/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreCliDriver.java +++ b/itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreCliDriver.java @@ -109,6 +109,7 @@ public void runTest(String testName, String fname, String fpath) { System.err.println("Begin query: " + fname); qt.setInputFile(fpath); + setupAdditionalPartialMasks(); qt.cliInit(); try { @@ -118,7 +119,6 @@ public void runTest(String testName, String fname, String fpath) { qt.failedQuery(e.getCause(), e.getResponseCode(), fname, QTestUtil.DEBUG_HINT); } - setupAdditionalPartialMasks(); QTestProcessExecResult result = qt.checkCliDriverResults(); resetAdditionalPartialMasks(); if (result.getReturnCode() != 0) { diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QOutProcessor.java b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QOutProcessor.java index 6198f3ea0bb1..5c8148a26886 100644 --- a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QOutProcessor.java +++ b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QOutProcessor.java @@ -18,12 +18,11 @@ package org.apache.hadoop.hive.ql; import java.io.BufferedReader; -import java.io.BufferedWriter; -import java.io.File; -import java.io.FileInputStream; -import java.io.FileOutputStream; -import java.io.InputStreamReader; -import java.io.OutputStreamWriter; +import java.io.IOException; +import java.io.StringReader; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -31,7 +30,6 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; -import org.apache.commons.io.FileUtils; import org.apache.commons.lang3.tuple.ImmutablePair; import org.apache.commons.lang3.tuple.Pair; import org.apache.hadoop.hive.ql.QTestMiniClusters.FsType; @@ -91,6 +89,11 @@ public String get() { } } + public static final class MaskingFoldState { + private boolean lastWasMasked; + private boolean lastWasVertexKilled; + } + private final Pattern[] planMask = toPattern(new String[] { ".*[.][.][.] [0-9]* more.*", "pk_-?[0-9]*_[0-9]*_[0-9]*", @@ -184,55 +187,73 @@ private Pattern[] toPattern(String[] patternStrs) { return patterns; } - public void maskPatterns(String fname) throws Exception { - - String line; - BufferedReader in; - BufferedWriter out; - - File file = new File(fname); - File fileOrig = new File(fname + ".orig"); - FileUtils.copyFile(file, fileOrig); - - in = new BufferedReader(new InputStreamReader(new FileInputStream(fileOrig), "UTF-8")); - out = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file), "UTF-8")); - - boolean lastWasMasked = false; - boolean lastWasVertexKilled = false; + /** + * Applies {@link #processLine(String)} and consecutive-line folding. + * + * @return the line to emit, or {@code null} if the line should be suppressed + */ + public String maskAndFoldLine(String line, MaskingFoldState state) { + LineProcessingResult result = processLine(line); + + if (result.line.equals(MASK_PATTERN)) { + // We're folding multiple masked lines into one. + if (state.lastWasMasked) { + state.lastWasVertexKilled = false; + return null; + } + state.lastWasMasked = true; + state.lastWasVertexKilled = false; + return result.line; + } + if (result.line.equals(MASKED_VERTEX_KILLED_PATTERN)) { + // Deduplicate consecutive standalone vertex-killed lines — the number of sibling + // vertices still alive when the kill propagates is non-deterministic. + if (state.lastWasVertexKilled) { + state.lastWasMasked = false; + return null; + } + state.lastWasVertexKilled = true; + state.lastWasMasked = false; + return result.line; + } + state.lastWasMasked = false; + state.lastWasVertexKilled = false; + return result.line; + } - while (null != (line = in.readLine())) { - LineProcessingResult result = processLine(line); + public List maskLines(List lines) { + MaskingFoldState state = new MaskingFoldState(); + List masked = new ArrayList<>(lines.size()); + for (String line : lines) { + String folded = maskAndFoldLine(line, state); + if (folded != null) { + masked.add(folded); + } + } + return masked; + } - if (result.line.equals(MASK_PATTERN)) { - // We're folding multiple masked lines into one. - if (!lastWasMasked) { - out.write(result.line); - out.write("\n"); - lastWasMasked = true; - result.partialMaskWasMatched = false; - } - lastWasVertexKilled = false; - } else if (result.line.equals(MASKED_VERTEX_KILLED_PATTERN)) { - // Deduplicate consecutive standalone vertex-killed lines — the number of sibling - // vertices still alive when the kill propagates is non-deterministic. - if (!lastWasVertexKilled) { - out.write(result.line); - out.write("\n"); - lastWasVertexKilled = true; - } - lastWasMasked = false; - result.partialMaskWasMatched = false; - } else { - out.write(result.line); - out.write("\n"); - lastWasMasked = false; - lastWasVertexKilled = false; - result.partialMaskWasMatched = false; + /** Applies masking and folding to multiline text */ + public String maskContent(String content) throws IOException { + List lines = new ArrayList<>(); + try (BufferedReader reader = new BufferedReader(new StringReader(content))) { + String line; + while ((line = reader.readLine()) != null) { + lines.add(line); } } + StringBuilder out = new StringBuilder(); + for (String line : maskLines(lines)) { + out.append(line).append('\n'); + } + return out.toString(); + } - in.close(); - out.close(); + /** Applies {@link #maskContent(String)} to a q test output file in place. */ + public void maskOutputFile(String fname) throws IOException { + Path path = Path.of(fname); + String masked = maskContent(Files.readString(path, StandardCharsets.UTF_8)); + Files.writeString(path, masked, StandardCharsets.UTF_8); } LineProcessingResult processLine(String line) { diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestResultProcessor.java b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestResultProcessor.java index eb7615b3c4e9..06b79a4fdead 100644 --- a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestResultProcessor.java +++ b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestResultProcessor.java @@ -93,7 +93,7 @@ public void setOutputs(CliSessionState ss, OutputStream fo) throws Exception { // specified below. This ensures the behavior remains the same as before this // refactoring. // It would be better to throw an error than silently pick one and ignore the - // rest but it is out of the scope of the current change. + // rest but it is out of the scope of the current change. if (operations.contains(Operation.SORT)) { ss.out = new SortPrintStream(fo, "UTF-8"); } else if (operations.contains(Operation.HASH)) { diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java index 447102f91c49..7024509b03ae 100644 --- a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java +++ b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java @@ -691,7 +691,7 @@ private void setSessionOutputs(CliSessionState ss, File outf) throws Exception { ss.out = new QTestFetchConverter(ss.out, false, "UTF-8", line -> { notifyOutputLine(line); if (qOutProcessor != null) { - // ensure that the masking is done before the sorting of the query results + // Mask before sort/hash (HIVE-29226); folding runs in maskOutputFile(). return qOutProcessor.processLine(line).get(); } return line; @@ -1014,7 +1014,8 @@ public QTestProcessExecResult checkCliDriverResults() throws Exception { String outFileName = outPath(outDir, tname + outFileExtension); File f = new File(logDir, tname + outFileExtension); - qOutProcessor.maskPatterns(f.getPath()); + + qOutProcessor.maskOutputFile(f.getPath()); if (QTestSystemProperties.shouldOverwriteResults()) { qTestResultProcessor.overwriteResults(f.getPath(), outFileName); diff --git a/itests/util/src/main/java/org/apache/hive/beeline/ConvertedOutputFile.java b/itests/util/src/main/java/org/apache/hive/beeline/ConvertedOutputFile.java index 2c5b1605c2c1..789915bd55a8 100644 --- a/itests/util/src/main/java/org/apache/hive/beeline/ConvertedOutputFile.java +++ b/itests/util/src/main/java/org/apache/hive/beeline/ConvertedOutputFile.java @@ -20,8 +20,10 @@ import org.apache.hadoop.hive.common.io.DigestPrintStream; import org.apache.hadoop.hive.common.io.FetchCallback; +import org.apache.hadoop.hive.common.io.QTestFetchConverter; import org.apache.hadoop.hive.common.io.SortAndDigestPrintStream; import org.apache.hadoop.hive.common.io.SortPrintStream; +import org.apache.hadoop.hive.ql.QOutProcessor; import java.io.PrintStream; @@ -33,10 +35,30 @@ public class ConvertedOutputFile extends OutputFile { private final boolean hasFetchCallback; public ConvertedOutputFile(OutputFile inner, Converter converter) throws Exception { - super(converter.getConvertedPrintStream(inner.getOut()), inner.getFilename()); + this(inner, converter, null, null); + } + + public ConvertedOutputFile(OutputFile inner, Converter converter, QOutProcessor qOutProcessor, + QOutProcessor.MaskingFoldState maskingFoldState) throws Exception { + super(wrapStream(inner.getOut(), converter, qOutProcessor, maskingFoldState), inner.getFilename()); hasFetchCallback = (getOut() instanceof FetchCallback); } + private static PrintStream wrapStream(PrintStream inner, Converter converter, + QOutProcessor qOutProcessor, QOutProcessor.MaskingFoldState maskingFoldState) throws Exception { + if (qOutProcessor == null || maskingFoldState == null) { + return converter.getConvertedPrintStream(inner); + } + // Sort before mask: identical MASK lines must not be created by sorting already-masked rows. + PrintStream masked = new QTestFetchConverter(inner, false, "UTF-8", line -> { + if (line.startsWith("Reading log file:")) { + return null; + } + return qOutProcessor.maskAndFoldLine(line, maskingFoldState); + }); + return converter.getConvertedPrintStream(masked); + } + @Override boolean isActiveConverter() { return hasFetchCallback; diff --git a/itests/util/src/main/java/org/apache/hive/beeline/QFile.java b/itests/util/src/main/java/org/apache/hive/beeline/QFile.java index aeb36f8f277e..56b0ab7068c3 100644 --- a/itests/util/src/main/java/org/apache/hive/beeline/QFile.java +++ b/itests/util/src/main/java/org/apache/hive/beeline/QFile.java @@ -331,26 +331,12 @@ public String filter(String input) { } } - // These are the filters which are common for every QTest. - // Check specificFilterSet for QTest specific ones. + // BeeLine-specific filters. Volatile-value masking is applied in-stream via QOutProcessor. private static RegexFilterSet getStaticFilterSet() { - // Pattern to remove the timestamp and other infrastructural info from the out file return new RegexFilterSet() .addFilter("Reading log file: .*\n", "") .addFilter("INFO : ", "") - .addFilter(".*/tmp/.*\n", MASK_PATTERN) - .addFilter(".*file:.*\n", MASK_PATTERN) - .addFilter(".*file\\..*\n", MASK_PATTERN) - .addFilter(".*Location.*\n", MASK_PATTERN) - .addFilter(".*LOCATION '.*\n", MASK_PATTERN) - .addFilter(".*Output:.*/data/files/.*\n", MASK_PATTERN) - .addFilter(".*CreateTime.*\n", MASK_PATTERN) - .addFilter(".*last_modified_.*\n", MASK_PATTERN) - .addFilter(".*transient_lastDdlTime.*\n", MASK_PATTERN) - .addFilter(".*lastUpdateTime.*\n", MASK_PATTERN) - .addFilter(".*lastAccessTime.*\n", MASK_PATTERN) - .addFilter(".*[Oo]wner.*\n", MASK_PATTERN) - .addFilter("(?s)(" + MASK_PATTERN + ")+", MASK_PATTERN); + .addFilter(".*Output:.*/data/files/.*\n", MASK_PATTERN); } /** diff --git a/itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java b/itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java index 2a1bb2cb2805..43442543576d 100644 --- a/itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java +++ b/itests/util/src/main/java/org/apache/hive/beeline/QFileBeeLineClient.java @@ -19,7 +19,7 @@ package org.apache.hive.beeline; import org.apache.commons.lang3.ArrayUtils; -import org.apache.hadoop.hive.ql.QTestUtil; +import org.apache.hadoop.hive.ql.QOutProcessor; import org.apache.hadoop.hive.ql.dataset.QTestDatasetHandler; import org.apache.hive.beeline.ConvertedOutputFile.Converter; @@ -43,6 +43,8 @@ public class QFileBeeLineClient implements AutoCloseable { private BeeLine beeLine; private PrintStream beelineOutputStream; private File logFile; + private QOutProcessor qOutProcessor; + private QOutProcessor.MaskingFoldState maskingFoldState; private String[] TEST_FIRST_COMMANDS = new String[] { "!set outputformat tsv2", "!set verbose false", @@ -126,13 +128,30 @@ private Set getViews() throws SQLException { return views; } + public void setOutputMasking(QOutProcessor qOutProcessor, + QOutProcessor.MaskingFoldState maskingFoldState) { + this.qOutProcessor = qOutProcessor; + this.maskingFoldState = maskingFoldState; + } + public void execute(String[] commands, File resultFile, Converter converter) throws Exception { + execute(commands, resultFile, converter, false); + } + + private void execute(String[] commands, File resultFile, Converter converter, boolean applyMasking) + throws Exception { beeLine.runCommands( new String[] { "!record " + resultFile.getAbsolutePath() }); - beeLine.setRecordOutputFile(new ConvertedOutputFile(beeLine.getRecordOutputFile(), converter)); + OutputFile recordOutputFile = beeLine.getRecordOutputFile(); + if (applyMasking) { + beeLine.setRecordOutputFile( + new ConvertedOutputFile(recordOutputFile, converter, qOutProcessor, maskingFoldState)); + } else { + beeLine.setRecordOutputFile(new ConvertedOutputFile(recordOutputFile, converter)); + } int lastSuccessfulCommand = beeLine.runCommands(commands); if (commands.length != lastSuccessfulCommand) { @@ -208,7 +227,7 @@ public void execute(QFile qFile, List> preCommands) throws Except } String[] commands = beeLine.getCommands(qFile.getInputFile()); - execute(qFile.filterCommands(commands), qFile.getRawOutputFile(), qFile.getConverter()); + execute(qFile.filterCommands(commands), qFile.getRawOutputFile(), qFile.getConverter(), true); afterExecute(qFile); } diff --git a/itests/util/src/test/java/org/apache/hadoop/hive/ql/TestQOutProcessor.java b/itests/util/src/test/java/org/apache/hadoop/hive/ql/TestQOutProcessor.java index 7bf4d5715cb7..8fc3ca76c429 100644 --- a/itests/util/src/test/java/org/apache/hadoop/hive/ql/TestQOutProcessor.java +++ b/itests/util/src/test/java/org/apache/hadoop/hive/ql/TestQOutProcessor.java @@ -17,8 +17,8 @@ */ package org.apache.hadoop.hive.ql; +import java.io.ByteArrayOutputStream; import java.io.File; -import java.io.PrintWriter; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.util.Arrays; @@ -26,13 +26,18 @@ import org.apache.hadoop.hive.ql.QTestMiniClusters.FsType; import org.apache.hadoop.hive.ql.qoption.QTestReplaceHandler; +import org.apache.hadoop.hive.common.io.QTestFetchConverter; +import org.apache.hadoop.hive.common.io.SortPrintStream; +import org.apache.hive.beeline.ConvertedOutputFile; +import org.apache.hive.beeline.ConvertedOutputFile.Converter; +import org.apache.hive.beeline.OutputFile; import org.junit.Assert; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; /** - * This class contains unit tests for QTestUtil + * Unit tests for {@link QOutProcessor} in-memory masking APIs and qtest capture streams. */ public class TestQOutProcessor { QOutProcessor qOutProcessor = new QOutProcessor(FsType.LOCAL, new QTestReplaceHandler()); @@ -40,6 +45,46 @@ public class TestQOutProcessor { @Rule public TemporaryFolder tmpFolder = new TemporaryFolder(); + /** + * Query fetch rows (-- SORT_QUERY_RESULTS) are one line per println; identity transform passes through. + */ + @Test + public void testSingleLineQueryRowPassThrough() throws Exception { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + QTestFetchConverter converter = new QTestFetchConverter(buf, true, "UTF-8", line -> line); + converter.println("0\tabc"); + Assert.assertEquals("0\tabc\n", buf.toString(StandardCharsets.UTF_8)); + } + + /** + * Transform returning null suppresses the line (BeeLine "Reading log file:" and mask folding). + */ + @Test + public void testNullTransformSuppressesLine() throws Exception { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + QTestFetchConverter converter = new QTestFetchConverter(buf, true, "UTF-8", + line -> line.startsWith("Reading log file:") ? null : line); + converter.println("Reading log file: /tmp/op.log"); + converter.println("PREHOOK: query: SELECT 1"); + Assert.assertEquals("PREHOOK: query: SELECT 1\n", buf.toString(StandardCharsets.UTF_8)); + } + + /** + * Consecutive full-mask lines fold to one via {@link QOutProcessor#maskLines(List)}. + */ + @Test + public void testConsecutiveMaskPatternFoldsInMemory() { + List input = Arrays.asList( + "line before", + QOutProcessor.MASK_PATTERN, + QOutProcessor.MASK_PATTERN, + QOutProcessor.MASK_PATTERN, + "line after"); + Assert.assertEquals( + Arrays.asList("line before", QOutProcessor.MASK_PATTERN, "line after"), + qOutProcessor.maskLines(input)); + } + /** * A raw vertex-killed log line must be replaced with MASKED_VERTEX_KILLED_PATTERN. */ @@ -103,46 +148,38 @@ public void testKilledVerticesCountIsMaskedInLongerLine() { /** * Multiple consecutive standalone MASKED_VERTEX_KILLED_PATTERN lines must be - * collapsed to a single line by maskPatterns(). + * collapsed to a single line by maskLines(). */ @Test - public void testConsecutiveVertexKilledLinesDeduplicatedInFile() throws Exception { - File f = tmpFile( + public void testConsecutiveVertexKilledLinesDeduplicatedInMemory() { + List input = Arrays.asList( "line before", QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, "line after"); - - qOutProcessor.maskPatterns(f.getAbsolutePath()); - - List lines = readLines(f); Assert.assertEquals( Arrays.asList("line before", QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, "line after"), - lines); + qOutProcessor.maskLines(input)); } /** * Two separate (non-consecutive) vertex-killed blocks must each produce one line. */ @Test - public void testNonConsecutiveVertexKilledLinesKeptSeparately() throws Exception { - File f = tmpFile( + public void testNonConsecutiveVertexKilledLinesKeptSeparatelyInMemory() { + List input = Arrays.asList( QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, "some other line", QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, QOutProcessor.MASKED_VERTEX_KILLED_PATTERN); - - qOutProcessor.maskPatterns(f.getAbsolutePath()); - - List lines = readLines(f); Assert.assertEquals( Arrays.asList( QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, "some other line", QOutProcessor.MASKED_VERTEX_KILLED_PATTERN), - lines); + qOutProcessor.maskLines(input)); } /** @@ -150,23 +187,178 @@ public void testNonConsecutiveVertexKilledLinesKeptSeparately() throws Exception * the run of vertex-killed lines. */ @Test - public void testVertexKilledRunResetByMaskedLine() throws Exception { - // "Deleted something" starts with "Deleted" → gets replaced by MASK_PATTERN - File f = tmpFile( + public void testVertexKilledRunResetByMaskedLineInMemory() { + List input = Arrays.asList( QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, - "Deleted /tmp/something", // will be masked → MASK_PATTERN + "Deleted /tmp/something", QOutProcessor.MASKED_VERTEX_KILLED_PATTERN); - - qOutProcessor.maskPatterns(f.getAbsolutePath()); - - List lines = readLines(f); - // MASK_PATTERN lines fold duplicates; but here there is only one occurrence Assert.assertEquals( Arrays.asList( QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, QOutProcessor.MASK_PATTERN, QOutProcessor.MASKED_VERTEX_KILLED_PATTERN), - lines); + qOutProcessor.maskLines(input)); + } + + @Test + public void testMaskContentFoldsConsecutiveVertexKilledLines() throws Exception { + String input = String.join("\n", + QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, + QOutProcessor.MASKED_VERTEX_KILLED_PATTERN, + "line after") + "\n"; + Assert.assertEquals( + QOutProcessor.MASKED_VERTEX_KILLED_PATTERN + "\nline after\n", + qOutProcessor.maskContent(input)); + } + + /** + * PREHOOK/POSTHOOK emit multiline query text in one println; stream masking must match + * {@link QOutProcessor#maskContent(String)} (including consecutive-line folding). + */ + @Test + public void testMultilinePrehookStreamMaskMatchesMaskContent() throws Exception { + String multiline = + "PREHOOK: query: CREATE TABLE alter_table_location2 (id int, name string, dept string)\n" + + " PARTITIONED BY (year int)\n" + + " STORED AS ORC\n" + + " LOCATION 'pfile://${system:test.tmp.dir}/alter_table_location2'\n" + + " TBLPROPERTIES (\"transactional\"=\"true\")"; + String expected = qOutProcessor.maskContent(multiline + "\n"); + + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); + QTestFetchConverter converter = new QTestFetchConverter(buf, true, "UTF-8", + line -> qOutProcessor.maskAndFoldLine(line, state)); + converter.println(multiline); + + Assert.assertEquals(expected, buf.toString(StandardCharsets.UTF_8)); + } + + @Test + public void testBeeLineRecordMaskingAppliedWhenProcessorProvided() throws Exception { + File out = tmpFolder.newFile("test.out"); + OutputFile inner = new OutputFile(out.getAbsolutePath()); + QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); + + ConvertedOutputFile convertedOutput = + new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); + convertedOutput.println("hdfs://localhost:51594/tmp/other text"); + convertedOutput.close(); + + String content = Files.readString(out.toPath(), StandardCharsets.UTF_8); + Assert.assertTrue(content.contains(QOutProcessor.HDFS_MASK)); + Assert.assertFalse(content.contains("localhost:51594")); + } + + /** + * BeeLine fetches HS2 operation logs in test mode; the "Reading log file:" header must be dropped + * (same as {@link org.apache.hive.beeline.QFile#getStaticFilterSet()}). + */ + @Test + public void testBeeLineRecordReadingLogFileLineIsSuppressed() throws Exception { + File out = tmpFolder.newFile("test.out"); + OutputFile inner = new OutputFile(out.getAbsolutePath()); + QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); + ConvertedOutputFile convertedOutput = + new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); + + convertedOutput.println("Reading log file: /tmp/cyanzheng/operation_logs/query.test"); + convertedOutput.println("PREHOOK: query: SELECT 1"); + convertedOutput.close(); + + Assert.assertEquals("PREHOOK: query: SELECT 1\n", + Files.readString(out.toPath(), StandardCharsets.UTF_8)); + } + + @Test + public void testBeeLineRecordNoMaskingWithoutProcessor() throws Exception { + File out = tmpFolder.newFile("test.out"); + OutputFile inner = new OutputFile(out.getAbsolutePath()); + + ConvertedOutputFile convertedOutput = new ConvertedOutputFile(inner, Converter.NONE); + convertedOutput.println("hdfs://localhost:51594/tmp/other text"); + convertedOutput.close(); + + Assert.assertEquals("hdfs://localhost:51594/tmp/other text\n", + Files.readString(out.toPath(), StandardCharsets.UTF_8)); + } + + /** + * BeeLine PREHOOK/POSTHOOK emit multiline query text in one println; converted output must match + * {@link QOutProcessor#maskContent(String)} via {@link ConvertedOutputFile}. + */ + @Test + public void testMultilinePrehookBeeLineRecordMaskMatchesMaskContent() throws Exception { + String multiline = + "PREHOOK: query: CREATE TABLE alter_table_location2 (id int, name string, dept string)\n" + + " PARTITIONED BY (year int)\n" + + " STORED AS ORC\n" + + " LOCATION 'pfile://${system:test.tmp.dir}/alter_table_location2'\n" + + " TBLPROPERTIES (\"transactional\"=\"true\")"; + String expected = qOutProcessor.maskContent(multiline + "\n"); + + File out = tmpFolder.newFile("test.out"); + OutputFile inner = new OutputFile(out.getAbsolutePath()); + QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); + ConvertedOutputFile convertedOutput = + new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); + + convertedOutput.println(multiline); + convertedOutput.close(); + + Assert.assertEquals(expected, Files.readString(out.toPath(), StandardCharsets.UTF_8)); + } + + /** + * With -- SORT_QUERY_RESULTS, sort unmasked rows first then apply masking on output so identical + * mask placeholders are not clustered by the sorter (same stream order as ConvertedOutputFile). + */ + @Test + public void testSortQueryResultsMasksAfterSort() throws Exception { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); + QTestFetchConverter mask = new QTestFetchConverter(buf, false, "UTF-8", + line -> qOutProcessor.maskAndFoldLine(line, state)); + SortPrintStream sort = new SortPrintStream(mask, "UTF-8"); + + sort.fetchStarted(); + sort.foundQuery(true); + sort.println("b\tcol\thdfs://host-b"); + sort.println("a\tcol\thdfs://host-a"); + sort.fetchFinished(); + + String output = buf.toString(StandardCharsets.UTF_8); + Assert.assertFalse(output.contains("host-a")); + Assert.assertFalse(output.contains("host-b")); + Assert.assertTrue(output.contains(QOutProcessor.HDFS_MASK)); + Assert.assertEquals( + "a\tcol\thdfs://" + QOutProcessor.HDFS_MASK + "\n" + + "b\tcol\thdfs://" + QOutProcessor.HDFS_MASK + "\n", + output); + } + + + @Test + public void testCliDriverMaskBeforeSort() throws Exception { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + SortPrintStream sort = new SortPrintStream(buf, "UTF-8"); + QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); + QTestFetchConverter mask = new QTestFetchConverter(sort, false, "UTF-8", + line -> qOutProcessor.maskAndFoldLine(line, state)); + + mask.fetchStarted(); + mask.foundQuery(true); + mask.println("b\tcol\thdfs://host-b"); + mask.println("a\tcol\thdfs://host-a"); + mask.fetchFinished(); + + String output = buf.toString(StandardCharsets.UTF_8); + Assert.assertFalse(output.contains("host-a")); + Assert.assertFalse(output.contains("host-b")); + Assert.assertEquals( + "a\tcol\thdfs://" + QOutProcessor.HDFS_MASK + "\n" + + "b\tcol\thdfs://" + QOutProcessor.HDFS_MASK + "\n", + output); } @Test @@ -219,22 +411,4 @@ private String processLine(String line) { return qOutProcessor.processLine(line).get(); } - private File tmpFile(String... lines) throws Exception { - File f = tmpFolder.newFile(); - try (PrintWriter pw = new PrintWriter(Files.newBufferedWriter(f.toPath(), StandardCharsets.UTF_8))) { - for (String l : lines) { - pw.println(l); - } - } - return f; - } - - private List readLines(File f) throws Exception { - List all = Files.readAllLines(f.toPath(), StandardCharsets.UTF_8); - while (!all.isEmpty() && all.get(all.size() - 1).isEmpty()) { - all.remove(all.size() - 1); - } - return all; - } - -} \ No newline at end of file +}