You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2023/12/02 10:00:07 UTC
(flink) 03/03: [FLINK-33600][table] print the time cost for batch query in the view
This is an automated email from the ASF dual-hosted git repository.
jingge pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c20c13fb5cb78eff2cbd08ea48f1cb7cf9a1981c
Author: jingge <ge...@gmail.com>
AuthorDate: Fri Nov 24 02:35:32 2023 +0100
[FLINK-33600][table] print the time cost for batch query in the view
---
.../table/client/cli/CliTableauResultView.java | 76 ++++++++++-
.../org/apache/flink/table/client/cli/Printer.java | 12 +-
.../client/cli/parser/SqlMultiLineParser.java | 5 +-
.../table/client/gateway/ResultDescriptor.java | 5 +
.../flink/table/client/cli/CliClientITCase.java | 4 +-
.../table/client/cli/CliTableauResultViewTest.java | 142 ++++++++++++++++++---
.../flink-sql-client/src/test/resources/sql/set.q | 3 +-
7 files changed, 220 insertions(+), 27 deletions(-)
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java
index 5338b4e603f..edf9c0c8f99 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java
@@ -38,6 +38,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
/** Print result in tableau mode. */
public class CliTableauResultView implements AutoCloseable {
@@ -48,21 +49,26 @@ public class CliTableauResultView implements AutoCloseable {
private final ChangelogResult collectResult;
private final ExecutorService displayResultExecutorService;
- public CliTableauResultView(final Terminal terminal, final ResultDescriptor resultDescriptor) {
- this(terminal, resultDescriptor, resultDescriptor.createResult());
+ private final long queryBeginTime;
+
+ public CliTableauResultView(
+ final Terminal terminal, final ResultDescriptor resultDescriptor, long queryBeginTime) {
+ this(terminal, resultDescriptor, resultDescriptor.createResult(), queryBeginTime);
}
@VisibleForTesting
public CliTableauResultView(
final Terminal terminal,
final ResultDescriptor resultDescriptor,
- final ChangelogResult collectResult) {
+ final ChangelogResult collectResult,
+ long queryBeginTime) {
this.terminal = terminal;
this.resultDescriptor = resultDescriptor;
this.collectResult = collectResult;
this.displayResultExecutorService =
Executors.newSingleThreadExecutor(
new ExecutorThreadFactory("CliTableauResultView"));
+ this.queryBeginTime = queryBeginTime;
}
public void displayResults() throws SqlExecutionException {
@@ -122,6 +128,27 @@ public class CliTableauResultView implements AutoCloseable {
private void printBatchResults(AtomicInteger receivedRowCount) {
final List<RowData> resultRows = waitBatchResults();
receivedRowCount.addAndGet(resultRows.size());
+
+ if (resultRows.isEmpty()) {
+ printBatchEmptySet();
+ } else {
+ printBatchTable(resultRows);
+ printBatchFooter(resultRows.size());
+ }
+ }
+
+ private void printBatchEmptySet() {
+ if (!resultDescriptor.isPrintQueryTimeCost()) {
+ terminal.writer().println("Empty set");
+ } else {
+ String timeCost =
+ calculateTimeCostInPrintFormat(queryBeginTime, System.currentTimeMillis());
+ terminal.writer().println("Empty set" + timeCost);
+ }
+ terminal.writer().flush();
+ }
+
+ private void printBatchTable(List<RowData> resultRows) {
TableauStyle style =
PrintStyle.tableauWithDataInferredColumnWidths(
resultDescriptor.getResultSchema(),
@@ -129,7 +156,39 @@ public class CliTableauResultView implements AutoCloseable {
resultDescriptor.maxColumnWidth(),
false,
false);
- style.print(resultRows.iterator(), terminal.writer());
+
+ List<String[]> content =
+ resultRows.stream().map(style::rowFieldsToString).collect(Collectors.toList());
+ // infer column width from the actual content
+ style.inferColumnWidth(content);
+ // print border line
+ style.printBorderLine(terminal.writer());
+ // print field names
+ style.printColumnNamesTableauRow(terminal.writer());
+ // print border line
+ style.printBorderLine(terminal.writer());
+
+ for (String[] resultRow : content) {
+ if (Thread.currentThread().isInterrupted()) {
+ return;
+ }
+ style.printTableauRow(resultRow, terminal.writer());
+ }
+
+ // print border line
+ style.printBorderLine(terminal.writer());
+ }
+
+ private void printBatchFooter(long numRows) {
+ final String rowTerm = numRows > 1 ? "rows" : "row";
+ if (!resultDescriptor.isPrintQueryTimeCost()) {
+ terminal.writer().println(numRows + " " + rowTerm + " in set");
+ } else {
+ String timeCost =
+ calculateTimeCostInPrintFormat(queryBeginTime, System.currentTimeMillis());
+ terminal.writer().println(numRows + " " + rowTerm + " in set" + timeCost);
+ }
+ terminal.writer().flush();
}
private void printStreamingResults(AtomicInteger receivedRowCount) {
@@ -212,4 +271,13 @@ public class CliTableauResultView implements AutoCloseable {
private String getRowTerm(AtomicInteger receivedRowCount) {
return receivedRowCount.get() > 1 ? "rows" : "row";
}
+
+ private String calculateTimeCostInPrintFormat(long queryBeginTime, long stopCountingTime) {
+ if (!resultDescriptor.isPrintQueryTimeCost()) {
+ return "";
+ }
+
+ String timeCost = String.format("%.2f", (stopCountingTime - queryBeginTime) / 1000d);
+ return " (" + timeCost + " seconds)";
+ }
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/Printer.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/Printer.java
index 54c1ca6d540..49209c5d750 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/Printer.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/Printer.java
@@ -64,8 +64,8 @@ public interface Printer extends Closeable {
}
static StatementResultPrinter createStatementCommandPrinter(
- StatementResult result, ReadableConfig sessionConfig) {
- return new StatementResultPrinter(result, sessionConfig);
+ StatementResult result, ReadableConfig sessionConfig, long startTime) {
+ return new StatementResultPrinter(result, sessionConfig, startTime);
}
static InitializationCommandPrinter createInitializationCommandPrinter() {
@@ -167,9 +167,13 @@ public interface Printer extends Closeable {
private final StatementResult result;
private final ReadableConfig sessionConfig;
- public StatementResultPrinter(StatementResult result, ReadableConfig sessionConfig) {
+ private final long queryBeginTime;
+
+ public StatementResultPrinter(
+ StatementResult result, ReadableConfig sessionConfig, long queryBeginTime) {
this.result = result;
this.sessionConfig = sessionConfig;
+ this.queryBeginTime = queryBeginTime;
}
@Override
@@ -192,7 +196,7 @@ public interface Printer extends Closeable {
final ResultDescriptor resultDesc = new ResultDescriptor(result, sessionConfig);
if (resultDesc.isTableauMode()) {
try (CliTableauResultView tableauResultView =
- new CliTableauResultView(terminal, resultDesc)) {
+ new CliTableauResultView(terminal, resultDesc, queryBeginTime)) {
tableauResultView.displayResults();
}
} else {
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlMultiLineParser.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlMultiLineParser.java
index 6de523e43dc..a1e73bbb95b 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlMultiLineParser.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlMultiLineParser.java
@@ -110,6 +110,7 @@ public class SqlMultiLineParser extends DefaultParser {
executor.configureSession(line);
printer = Printer.createInitializationCommandPrinter();
} else {
+ long queryBeginTime = System.currentTimeMillis();
StatementResult result = executor.executeStatement(line);
ReadableConfig sessionConfig = executor.getSessionConfig();
if (mode == CliClient.ExecutionMode.NON_INTERACTIVE_EXECUTION
@@ -125,7 +126,9 @@ public class SqlMultiLineParser extends DefaultParser {
EXECUTION_RESULT_MODE.key(),
TABLEAU));
}
- printer = Printer.createStatementCommandPrinter(result, sessionConfig);
+ printer =
+ Printer.createStatementCommandPrinter(
+ result, sessionConfig, queryBeginTime);
}
break;
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java
index 33543de9619..d1c5c6dd66e 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java
@@ -32,6 +32,7 @@ import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
import static org.apache.flink.table.client.config.ResultMode.CHANGELOG;
import static org.apache.flink.table.client.config.ResultMode.TABLE;
import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_MAX_COLUMN_WIDTH;
+import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_QUERY_TIME_COST;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_MAX_TABLE_RESULT_ROWS;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_RESULT_MODE;
@@ -94,6 +95,10 @@ public class ResultDescriptor {
return config.get(DISPLAY_MAX_COLUMN_WIDTH);
}
+ public boolean isPrintQueryTimeCost() {
+ return config.get(DISPLAY_QUERY_TIME_COST);
+ }
+
public RowDataToStringConverter getRowDataStringConverter() {
return tableResult.getRowDataToStringConverter();
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java
index c23c1e8b847..1cb37239b5b 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.client.cli.parser.SqlCommandParserImpl;
import org.apache.flink.table.client.cli.parser.SqlMultiLineParser;
import org.apache.flink.table.client.cli.utils.SqlScriptReader;
import org.apache.flink.table.client.cli.utils.TestSqlStatement;
+import org.apache.flink.table.client.config.SqlClientOptions;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.SingleSessionManager;
import org.apache.flink.table.data.GenericRowData;
@@ -271,7 +272,8 @@ class CliClientITCase {
// Make sure we use the new cast behaviour
.set(
ExecutionConfigOptions.TABLE_EXEC_LEGACY_CAST_BEHAVIOUR,
- ExecutionConfigOptions.LegacyCastBehaviour.DISABLED),
+ ExecutionConfigOptions.LegacyCastBehaviour.DISABLED)
+ .set(SqlClientOptions.DISPLAY_QUERY_TIME_COST, false),
Collections.emptyList());
InputStream inputStream = new ByteArrayInputStream(sqlContent.getBytes());
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
index b17861f8f6c..e4922ae4cdf 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
@@ -59,6 +59,7 @@ import java.util.stream.Collectors;
import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_MAX_COLUMN_WIDTH;
+import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_QUERY_TIME_COST;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_RESULT_MODE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -180,15 +181,20 @@ class CliTableauResultViewTest {
}
@Test
- void testBatchResult() {
+ void testBatchResultWithDisabledDisplayTimeCost() {
final Configuration testConfig = new Configuration();
testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ testConfig.set(DISPLAY_QUERY_TIME_COST, false);
ResultDescriptor resultDescriptor =
new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, createTestChangelogResult());
+ new CliTableauResultView(
+ terminal,
+ resultDescriptor,
+ createNewTestChangelogResult(),
+ System.currentTimeMillis());
view.displayResults();
assertThat(terminalOutput)
.hasToString(
@@ -219,14 +225,69 @@ class CliTableauResultViewTest {
+ "8 rows in set"
+ System.lineSeparator());
+ view.close();
+
// adjust the max column width for printing
testConfig.set(DISPLAY_MAX_COLUMN_WIDTH, 80);
- TestChangelogResult collectResult = createTestChangelogResult();
- view = new CliTableauResultView(terminal, resultDescriptor, collectResult);
+ TestChangelogResult collectResult = createNewTestChangelogResult();
+ view =
+ new CliTableauResultView(
+ terminal, resultDescriptor, collectResult, System.currentTimeMillis());
view.displayResults();
assertThat(terminalOutput.toString()).contains("abcdefghijklmnopqrstuvwxyz12345");
+ view.close();
+
+ // Job is finished. Don't need to close the job manually.
+ assertThat(collectResult.closed).isFalse();
+ }
+
+ @Test
+ void testBatchResultWithDisplayTimeCost() {
+ final Configuration testConfig = new Configuration();
+ testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
+ testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+
+ ResultDescriptor resultDescriptor =
+ new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
+ TestChangelogResult collectResult = createNewTestChangelogResult();
+ CliTableauResultView view =
+ new CliTableauResultView(
+ terminal, resultDescriptor, collectResult, System.currentTimeMillis());
+ view.displayResults();
+ assertThat(terminalOutput.toString())
+ .contains(
+ "+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+---------------------+"
+ + System.lineSeparator()
+ + "| boolean | int | bigint | varchar | decimal(10, 5) | timestamp | binary |"
+ + System.lineSeparator()
+ + "+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+---------------------+"
+ + System.lineSeparator()
+ + "| <NULL> | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 | x'32333485365d737e' |"
+ + System.lineSeparator()
+ + "| FALSE | <NULL> | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 | x'649e207983' |"
+ + System.lineSeparator()
+ + "| TRUE | 2147483647 | <NULL> | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 | x'92e90102' |"
+ + System.lineSeparator()
+ + "| FALSE | -2147483648 | 9223372036854775807 | <NULL> | 12345.06789 | 2020-03-01 18:39:14.123000 | x'32333485365d737e' |"
+ + System.lineSeparator()
+ + "| TRUE | 100 | -9223372036854775808 | abcdefg111 | <NULL> | 2020-03-01 18:39:14.123456 | x'6e17fffe' |"
+ + System.lineSeparator()
+ + "| <NULL> | -1 | -1 | abcdefghijklmnopqrstuvwxyz1... | -12345.06789 | <NULL> | <NULL> |"
+ + System.lineSeparator()
+ + "| <NULL> | -1 | -1 | 这是一段中文 | -12345.06789 | 2020-03-04 18:39:14.000000 | x'fdfeff00010203' |"
+ + System.lineSeparator()
+ + "| <NULL> | -1 | -1 | これは日本語をテストするた... | -12345.06789 | 2020-03-04 18:39:14.000000 | x'fdfeff00010203' |"
+ + System.lineSeparator()
+ + "+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+---------------------+"
+ + System.lineSeparator()
+ + "8 rows in set");
+
+ String[] outputLines = terminalOutput.toString().split("\\r?\\n");
+ assertThat(outputLines[outputLines.length - 1])
+ .matches("8 rows in set \\(\\d+\\.\\d{2} seconds\\)");
+
view.close();
// Job is finished. Don't need to close the job manually.
assertThat(collectResult.closed).isFalse();
@@ -237,6 +298,8 @@ class CliTableauResultViewTest {
final Configuration testConfig = new Configuration();
testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ testConfig.set(DISPLAY_QUERY_TIME_COST, false);
+
ResultDescriptor resultDescriptor =
new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
TestChangelogResult collectResult =
@@ -244,7 +307,8 @@ class CliTableauResultViewTest {
() -> TypedResult.payload(data.subList(0, data.size() / 2)),
TypedResult::empty);
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, collectResult);
+ new CliTableauResultView(
+ terminal, resultDescriptor, collectResult, System.currentTimeMillis());
// submit result display in another thread
Future<?> furture = EXECUTOR_RESOURCE.getExecutor().submit(view::displayResults);
@@ -265,21 +329,54 @@ class CliTableauResultViewTest {
}
@Test
- void testEmptyBatchResult() {
+ void testEmptyBatchResultWithDisabledDisplayTimeCost() {
final Configuration testConfig = new Configuration();
testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ testConfig.set(DISPLAY_QUERY_TIME_COST, false);
ResultDescriptor resultDescriptor =
new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
TestChangelogResult testChangelogResult = new TestChangelogResult(TypedResult::endOfStream);
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, testChangelogResult);
+ new CliTableauResultView(
+ terminal,
+ resultDescriptor,
+ testChangelogResult,
+ System.currentTimeMillis());
view.displayResults();
view.close();
assertThat(terminalOutput).hasToString("Empty set" + System.lineSeparator());
+
+ // Job is finished. Don't need to close the job manually.
+ assertThat(testChangelogResult.closed).isFalse();
+ }
+
+ @Test
+ void testEmptyBatchResultWithDisplayingTimeCost() {
+ final Configuration testConfig = new Configuration();
+ testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
+ testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+
+ ResultDescriptor resultDescriptor =
+ new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
+ TestChangelogResult testChangelogResult = new TestChangelogResult(TypedResult::endOfStream);
+ CliTableauResultView view =
+ new CliTableauResultView(
+ terminal,
+ resultDescriptor,
+ testChangelogResult,
+ System.currentTimeMillis());
+
+ view.displayResults();
+ view.close();
+
+ String[] outputLines = terminalOutput.toString().split("\\r?\\n");
+ assertThat(outputLines[outputLines.length - 1])
+ .matches("Empty set \\(\\d+\\.\\d{2} seconds\\)");
+
// Job is finished. Don't need to close the job manually.
assertThat(testChangelogResult.closed).isFalse();
}
@@ -298,7 +395,8 @@ class CliTableauResultViewTest {
});
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, changelogResult);
+ new CliTableauResultView(
+ terminal, resultDescriptor, changelogResult, System.currentTimeMillis());
assertThatThrownBy(view::displayResults)
.satisfies(anyCauseMatches(SqlExecutionException.class, "query failed"));
@@ -311,12 +409,14 @@ class CliTableauResultViewTest {
final Configuration testConfig = new Configuration();
testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.STREAMING);
+ testConfig.set(DISPLAY_QUERY_TIME_COST, false);
ResultDescriptor resultDescriptor =
new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
- TestChangelogResult collectResult = createTestChangelogResult();
+ TestChangelogResult collectResult = createNewTestChangelogResult();
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, collectResult);
+ new CliTableauResultView(
+ terminal, resultDescriptor, collectResult, System.currentTimeMillis());
view.displayResults();
view.close();
@@ -357,8 +457,10 @@ class CliTableauResultViewTest {
// adjust the max column width for printing
testConfig.set(DISPLAY_MAX_COLUMN_WIDTH, 80);
- collectResult = createTestChangelogResult();
- view = new CliTableauResultView(terminal, resultDescriptor, collectResult);
+ collectResult = createNewTestChangelogResult();
+ view =
+ new CliTableauResultView(
+ terminal, resultDescriptor, collectResult, System.currentTimeMillis());
view.displayResults();
assertThat(terminalOutput.toString()).contains("abcdefghijklmnopqrstuvwxyz12345");
@@ -371,12 +473,15 @@ class CliTableauResultViewTest {
final Configuration testConfig = new Configuration();
testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.STREAMING);
+ testConfig.set(DISPLAY_QUERY_TIME_COST, false);
+
ResultDescriptor resultDescriptor =
new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
TestChangelogResult collectResult = new TestChangelogResult(TypedResult::endOfStream);
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, collectResult);
+ new CliTableauResultView(
+ terminal, resultDescriptor, collectResult, System.currentTimeMillis());
view.displayResults();
view.close();
@@ -391,6 +496,7 @@ class CliTableauResultViewTest {
+ System.lineSeparator()
+ "Received a total of 0 row"
+ System.lineSeparator());
+
// Job is finished. Don't need to close the job manually.
assertThat(collectResult.closed).isFalse();
}
@@ -400,6 +506,7 @@ class CliTableauResultViewTest {
final Configuration testConfig = new Configuration();
testConfig.set(EXECUTION_RESULT_MODE, ResultMode.TABLEAU);
testConfig.set(RUNTIME_MODE, RuntimeExecutionMode.STREAMING);
+ testConfig.set(DISPLAY_QUERY_TIME_COST, false);
TestChangelogResult collectResult =
new TestChangelogResult(
@@ -410,7 +517,8 @@ class CliTableauResultViewTest {
ResultDescriptor resultDescriptor =
new ResultDescriptor(CliClientTestUtils.createTestClient(schema), testConfig);
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, collectResult);
+ new CliTableauResultView(
+ terminal, resultDescriptor, collectResult, System.currentTimeMillis());
// submit result display in another thread
Future<?> furture = EXECUTOR_RESOURCE.getExecutor().submit(view::displayResults);
@@ -441,6 +549,7 @@ class CliTableauResultViewTest {
+ System.lineSeparator()
+ "Query terminated, received a total of 4 rows"
+ System.lineSeparator());
+
// close job manually
assertThat(collectResult.closed).isTrue();
}
@@ -458,7 +567,8 @@ class CliTableauResultViewTest {
throw new SqlExecutionException("query failed");
});
CliTableauResultView view =
- new CliTableauResultView(terminal, resultDescriptor, changelogResult);
+ new CliTableauResultView(
+ terminal, resultDescriptor, changelogResult, System.currentTimeMillis());
assertThatThrownBy(view::displayResults)
.satisfies(anyCauseMatches(SqlExecutionException.class, "query failed"));
@@ -467,7 +577,7 @@ class CliTableauResultViewTest {
}
@NotNull
- private TestChangelogResult createTestChangelogResult() {
+ private TestChangelogResult createNewTestChangelogResult() {
TestChangelogResult collectResult =
new TestChangelogResult(
() -> TypedResult.payload(data.subList(0, data.size() / 2)),
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/set.q b/flink-table/flink-sql-client/src/test/resources/sql/set.q
index b7cc6855a1d..d55dfd8b1ab 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/set.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/set.q
@@ -86,10 +86,11 @@ set;
| pipeline.classpaths | |
| pipeline.jars | |
| rest.port | $VAR_REST_PORT |
+| sql-client.display.print-time-cost | false |
| sql-client.execution.result-mode | tableau |
| table.exec.legacy-cast-behaviour | DISABLED |
+--------------------------------------------+-----------+
-11 rows in set
+12 rows in set
!ok
# reset the configuration