From 11dbfbf4fb88cfdb9157ca9b0a4bb8c08b817ad8 Mon Sep 17 00:00:00 2001 From: cyan-zheng Date: Thu, 23 Jul 2026 16:06:02 +1000 Subject: [PATCH 1/2] HIVE-29227: Apply masking in memory instead of rewriting the file in QOutProcessor#maskPatterns Refactored masking from QOutProcessor#maskPatterns that involves file rewrite to in-memory implementation for CliDriver and BeeLine. --- .../hadoop/hive/common/io/FetchConverter.java | 11 +- .../hive/common/io/QTestFetchConverter.java | 21 +- .../hive/cli/control/CoreBeeLineDriver.java | 35 ++- .../hive/cli/control/CoreCliDriver.java | 2 +- .../apache/hadoop/hive/ql/QOutProcessor.java | 115 ++++---- .../org/apache/hadoop/hive/ql/QTestUtil.java | 13 +- .../hive/beeline/ConvertedOutputFile.java | 24 +- .../java/org/apache/hive/beeline/QFile.java | 18 +- .../hive/beeline/QFileBeeLineClient.java | 25 +- .../hadoop/hive/ql/TestQOutProcessor.java | 266 +++++++++++++++--- 10 files changed, 405 insertions(+), 125 deletions(-) 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..cc6c2ac47bd2 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,26 @@ 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. + if (str.indexOf('\n') < 0) { + writeTransformedLine(str); + return; + } + 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)); + } + + 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..ce12554b9d9a 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,8 @@ 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.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -31,7 +27,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 +86,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 +184,66 @@ 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; - - while (null != (line = in.readLine())) { - LineProcessingResult result = processLine(line); + /** + * 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; + } - 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; + 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; + } - in.close(); - out.close(); + /** 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(); } LineProcessingResult processLine(String line) { 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..42076dcbdfc5 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 @@ -143,6 +143,8 @@ public class QTestUtil { QTestOptionDispatcher dispatcher = new QTestOptionDispatcher(); private boolean isSessionStateStarted = false; + private QOutProcessor.MaskingFoldState outMaskingFoldState; + private QOutProcessor.MaskingFoldState errMaskingFoldState; public CliDriver getCliDriver() { if (cliDriver == null) { @@ -688,16 +690,24 @@ private void setSessionOutputs(CliSessionState ss, File outf) throws Exception { qTestResultProcessor.setOutputs(ss, fo); + outMaskingFoldState = new QOutProcessor.MaskingFoldState(); 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 - return qOutProcessor.processLine(line).get(); + return qOutProcessor.maskAndFoldLine(line, outMaskingFoldState); } return line; }); + errMaskingFoldState = new QOutProcessor.MaskingFoldState(); ss.err = new CachingPrintStream(fo, true, "UTF-8"); + ss.err = new QTestFetchConverter(ss.err, false, "UTF-8", line ->{ + if (qOutProcessor != null) { + return qOutProcessor.maskAndFoldLine(line, errMaskingFoldState); + } + return line; + }); ss.setIsSilent(true); ss.setIsQtestLogging(true); } @@ -1014,7 +1024,6 @@ public QTestProcessExecResult checkCliDriverResults() throws Exception { String outFileName = outPath(outDir, tname + outFileExtension); File f = new File(logDir, tname + outFileExtension); - qOutProcessor.maskPatterns(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..1b834b43bf19 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 record = + new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); + record.println("hdfs://localhost:51594/tmp/other text"); + record.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 record = + new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); + + record.println("Reading log file: /tmp/cyanzheng/operation_logs/query.test"); + record.println("PREHOOK: query: SELECT 1"); + record.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 record = new ConvertedOutputFile(inner, Converter.NONE); + record.println("hdfs://localhost:51594/tmp/other text"); + record.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; !record 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 record = + new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); + + record.println(multiline); + record.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 +} From 3ccc5e088dd41b65c2c7aa96c3f98ada241c0320 Mon Sep 17 00:00:00 2001 From: cyan-zheng Date: Tue, 28 Jul 2026 16:44:03 +1000 Subject: [PATCH 2/2] HIVE-29227: Fix CliDriver masking to preserve sort order and golden diffs Apply processLine on stdout before SORT_QUERY_RESULTS (HIVE-29226) and run maskAndFoldLine via maskOutputFile() before diff so stderr folding matches master. Keep CachingPrintStream as the outer stderr wrapper for hook tests, and skip multiline splitting in QTestFetchConverter during active query fetch. --- .../hive/common/io/CachingPrintStream.java | 8 +++++- .../hive/common/io/QTestFetchConverter.java | 28 ++++++++++++------- .../apache/hadoop/hive/ql/QOutProcessor.java | 10 +++++++ .../hadoop/hive/ql/QTestResultProcessor.java | 2 +- .../org/apache/hadoop/hive/ql/QTestUtil.java | 16 +++-------- .../hadoop/hive/ql/TestQOutProcessor.java | 28 +++++++++---------- 6 files changed, 54 insertions(+), 38 deletions(-) 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/QTestFetchConverter.java b/common/src/java/org/apache/hadoop/hive/common/io/QTestFetchConverter.java index cc6c2ac47bd2..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,19 +43,27 @@ public QTestFetchConverter(OutputStream out, boolean autoFlush, String encoding, @Override public void println(String str) { - // PREHOOK/POSTHOOK may send multiple lines separated by \n in one println, mask each separately. - if (str.indexOf('\n') < 0) { - writeTransformedLine(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; } - 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); + } + + private boolean shouldSplitMultiline() { + if (inner instanceof FetchConverter fetchConverter) { + return !(fetchConverter.queryfound && fetchConverter.fetchStarted); } - writeTransformedLine(str.substring(start)); + return true; } private void writeTransformedLine(String line) { 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 ce12554b9d9a..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 @@ -20,6 +20,9 @@ import java.io.BufferedReader; 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; @@ -246,6 +249,13 @@ public String maskContent(String content) throws IOException { return out.toString(); } + /** 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) { LineProcessingResult result = new LineProcessingResult(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 42076dcbdfc5..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 @@ -143,8 +143,6 @@ public class QTestUtil { QTestOptionDispatcher dispatcher = new QTestOptionDispatcher(); private boolean isSessionStateStarted = false; - private QOutProcessor.MaskingFoldState outMaskingFoldState; - private QOutProcessor.MaskingFoldState errMaskingFoldState; public CliDriver getCliDriver() { if (cliDriver == null) { @@ -690,24 +688,16 @@ private void setSessionOutputs(CliSessionState ss, File outf) throws Exception { qTestResultProcessor.setOutputs(ss, fo); - outMaskingFoldState = new QOutProcessor.MaskingFoldState(); 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 - return qOutProcessor.maskAndFoldLine(line, outMaskingFoldState); + // Mask before sort/hash (HIVE-29226); folding runs in maskOutputFile(). + return qOutProcessor.processLine(line).get(); } return line; }); - errMaskingFoldState = new QOutProcessor.MaskingFoldState(); ss.err = new CachingPrintStream(fo, true, "UTF-8"); - ss.err = new QTestFetchConverter(ss.err, false, "UTF-8", line ->{ - if (qOutProcessor != null) { - return qOutProcessor.maskAndFoldLine(line, errMaskingFoldState); - } - return line; - }); ss.setIsSilent(true); ss.setIsQtestLogging(true); } @@ -1025,6 +1015,8 @@ public QTestProcessExecResult checkCliDriverResults() throws Exception { File f = new File(logDir, tname + outFileExtension); + qOutProcessor.maskOutputFile(f.getPath()); + if (QTestSystemProperties.shouldOverwriteResults()) { qTestResultProcessor.overwriteResults(f.getPath(), outFileName); return QTestProcessExecResult.createWithoutOutput(0); 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 1b834b43bf19..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 @@ -240,10 +240,10 @@ public void testBeeLineRecordMaskingAppliedWhenProcessorProvided() throws Except OutputFile inner = new OutputFile(out.getAbsolutePath()); QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); - ConvertedOutputFile record = + ConvertedOutputFile convertedOutput = new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); - record.println("hdfs://localhost:51594/tmp/other text"); - record.close(); + 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)); @@ -259,12 +259,12 @@ public void testBeeLineRecordReadingLogFileLineIsSuppressed() throws Exception { File out = tmpFolder.newFile("test.out"); OutputFile inner = new OutputFile(out.getAbsolutePath()); QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); - ConvertedOutputFile record = + ConvertedOutputFile convertedOutput = new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); - record.println("Reading log file: /tmp/cyanzheng/operation_logs/query.test"); - record.println("PREHOOK: query: SELECT 1"); - record.close(); + 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)); @@ -275,16 +275,16 @@ public void testBeeLineRecordNoMaskingWithoutProcessor() throws Exception { File out = tmpFolder.newFile("test.out"); OutputFile inner = new OutputFile(out.getAbsolutePath()); - ConvertedOutputFile record = new ConvertedOutputFile(inner, Converter.NONE); - record.println("hdfs://localhost:51594/tmp/other text"); - record.close(); + 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; !record output must match + * BeeLine PREHOOK/POSTHOOK emit multiline query text in one println; converted output must match * {@link QOutProcessor#maskContent(String)} via {@link ConvertedOutputFile}. */ @Test @@ -300,11 +300,11 @@ public void testMultilinePrehookBeeLineRecordMaskMatchesMaskContent() throws Exc File out = tmpFolder.newFile("test.out"); OutputFile inner = new OutputFile(out.getAbsolutePath()); QOutProcessor.MaskingFoldState state = new QOutProcessor.MaskingFoldState(); - ConvertedOutputFile record = + ConvertedOutputFile convertedOutput = new ConvertedOutputFile(inner, Converter.NONE, qOutProcessor, state); - record.println(multiline); - record.close(); + convertedOutput.println(multiline); + convertedOutput.close(); Assert.assertEquals(expected, Files.readString(out.toPath(), StandardCharsets.UTF_8)); }