thomasrebele commented on code in PR #6632:
URL: https://github.com/apache/hive/pull/6632#discussion_r3911978659
##########
itests/util/src/main/java/org/apache/hadoop/hive/cli/control/CoreBeeLineDriver.java:
##########
@@ -260,13 +266,40 @@ private void runTest(QFile qFile, List<Callable<Void>>
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() {
Review Comment:
This method is almost a copy of CoreCliDriver#setupAdditionalPartialMasks.
Would it be possible to combine these two?
##########
itests/util/src/main/java/org/apache/hive/beeline/ConvertedOutputFile.java:
##########
@@ -34,10 +36,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.
Review Comment:
I don't see the relationship with sorting here. Could you explain it, please?
##########
itests/util/src/main/java/org/apache/hadoop/hive/ql/QOutProcessor.java:
##########
@@ -185,55 +188,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<String> maskLines(List<String> lines) {
+ MaskingFoldState state = new MaskingFoldState();
+ List<String> masked = new ArrayList<>(lines.size());
+ for (String line : lines) {
+ String folded = maskAndFoldLine(line, state);
+ if (folded != null) {
+ masked.add(folded);
+ }
+ }
+ return masked;
+ }
Review Comment:
The logic of this method can be integrated into the function
`maskContent(String content)`. That way a list creation can be avoided.
##########
itests/util/src/main/java/org/apache/hive/beeline/ConvertedOutputFile.java:
##########
@@ -34,10 +36,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:")) {
Review Comment:
This line feels a bit out of place here. I would not expect the
`ConvertedOutputFile` to define its own filters. I guess this corresponds to
org.apache.hive.beeline.QFile#getStaticFilterSet. (That method defines also two
other patterns. I'm wondering why `"INFO : "`, the other pattern that is
replaced with the empty string, is not necessary here.)
##########
common/src/java/org/apache/hadoop/hive/common/io/FetchConverter.java:
##########
@@ -49,8 +50,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);
+ }
Review Comment:
See [my comment on
CachingPrintStream](https://github.com/apache/hive/pull/6632/changes#r3913280319).
##########
common/src/java/org/apache/hadoop/hive/common/io/CachingPrintStream.java:
##########
@@ -44,7 +45,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);
+ }
Review Comment:
I would prefer to remove that change, as it complicates the IO logic even
further. Is the change actually necessary? I've commented out this change and
the one from FetchConverter.java and ran the failed tests (see script below).
Only cachingprintstream.q.out changed, which is the same behavior when I
execute the tests on the base commit of the PR
(3df58f1bef47b64f24ccf95eb75c67d261438a7f).
Here the script:
```
mvn test -Pitests -pl itests/qtest -Dtest=TestCliDriver
-Dqfile=inputwherefalse.q,stats_noscan_2.q -Dtest.output.overwrite=true
mvn test -Pitests -pl itests/qtest -Dtest=TestHBaseCliDriver
-Dqfile=hbase_bulk.q,hbase_handler_bulk.q -Dtest.output.overwrite=true
mvn test -Pitests -pl itests/qtest -Dtest=TestIcebergCliDriver
-Dqfile=query_iceberg_metadata_of_unpartitioned_table.q
-Dtest.output.overwrite=true
mvn test -Pitests -pl itests/qtest
-Dtest=TestIcebergLlapLocalCompactorCliDriver
-Dqfile=iceberg_major_compaction_query_metadata.q -Dtest.output.overwrite=true
mvn test -Pitests -pl itests/qtest -Dtest=TestMiniLlapCliDriver
-Dqfile=newline.q,remote_script.q -Dtest.output.overwrite=true
mvn test -Pitests -pl itests/qtest -Dtest=TestMiniLlapLocalCliDriver
-Dqfile=acid_stats4.q,alter_merge_3.q,alter_partition_format_loc.q,bucket_if_with_path_filter.q,cli_print_escape_crlf.q,create_like.q,create_table_like_file.q,create_table_like_file_orc.q,database_drop.q,database_location.q,describe_database_json.q,drop_database_removes_partition_dirs.q,drop_table_removes_partition_dirs.q,escape_crlf.q,exim_10_external_managed.q,exim_12_external_location.q,exim_13_managed_location.q,exim_14_managed_location_over_existing.q,exim_15_external_part.q,exim_16_part_external.q,exim_17_part_managed.q,exim_19_00_part_external_location.q,exim_19_part_external_location.q,exim_20_part_managed_location.q,external_jdbc_auth.q,insert_overwrite_directory.q,insert_overwrite_directory2.q,insertexternal1.q,load_data_using_job.q,load_fs.q,load_orc.q,load_orc_part.q,masking_8.q,nullformatdir.q,ppd_multi_insert.q,rfc5424_parser.q,rfc5424_parser_exception.q,sfs.q,sysdb.q,table_storage.q,udf_get_json_ob
ject.q,vectorization_escape_crlf.q,virtual_column.q
-Dtest.output.overwrite=true
mvn test -Pitests -pl itests/qtest -Dtest=TestNegativeLlapLocalCliDriver
-Dqfile=authorization_uri_export.q,cachingprintstream.q,exim_20_managed_location_over_existing.q,load_orc_negative_part.q
-Dtest.output.overwrite=true
```
##########
itests/util/src/test/java/org/apache/hadoop/hive/ql/TestQOutProcessor.java:
##########
@@ -104,70 +149,217 @@ 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<String> 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<String> 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<String> 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<String> lines = readLines(f);
Assert.assertEquals(
Arrays.asList(
QOutProcessor.MASKED_VERTEX_KILLED_PATTERN,
"some other line",
QOutProcessor.MASKED_VERTEX_KILLED_PATTERN),
- lines);
+ qOutProcessor.maskLines(input));
}
/**
* Vertex-killed deduplication must reset when a normal (masked) line
interrupts
* 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<String> 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<String> 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 {
Review Comment:
I don't see the purpose of the test. The asserts are the same as
testCliDriverMaskBeforeSort, so the test method does not verify what its name
advertises. Also [HIVE-29226](https://issues.apache.org/jira/browse/HIVE-29226)
changed SORT_QUERY_RESULTS to apply the masking before sorting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
