This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 527e676d137 [FLINK-24908][sql-client] Print the outermost exception
that ends with the root cause
527e676d137 is described below
commit 527e676d137301f5b0d10ee9fec895fb9fa2f43e
Author: Martijn Visser <[email protected]>
AuthorDate: Mon Sep 28 22:09:52 2026 +0200
[FLINK-24908][sql-client] Print the outermost exception that ends with the
root cause
The client printed the deepest cause, a Calcite class with the bare
text, so the position the planner now adds never reached the terminal.
It now prints the outermost exception whose message ends with the root
cause's message, in the chain and in gateway stack traces. The planner's
parse exception is picked by type, since it echoes the line inside the
Calcite text.
Generated-by: Claude Code (Claude Opus 5.5, Claude Fable 5.1)
---
.../apache/flink/table/client/cli/CliStrings.java | 130 +++++++-
.../flink/table/client/gateway/ExecutorImpl.java | 12 +-
.../flink/table/client/cli/CliStringsTest.java | 344 +++++++++++++++++++++
.../table/client/gateway/ExecutorImplITCase.java | 30 ++
.../src/test/resources/sql/catalog_database.q | 6 +-
.../src/test/resources/sql/module.q | 14 +-
.../src/test/resources/sql/select.q | 32 ++
.../flink-sql-client/src/test/resources/sql/set.q | 4 +-
.../src/test/resources/sql/table.q | 2 +-
.../src/test/resources/sql_multi/statement_set.q | 16 +-
10 files changed, 560 insertions(+), 30 deletions(-)
diff --git
a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
index 1287ac4d7ca..0ce8a027763 100644
---
a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
+++
b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.client.cli;
import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.table.api.SqlParserException;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
@@ -27,12 +28,37 @@ import org.jline.utils.AttributedString;
import org.jline.utils.AttributedStringBuilder;
import org.jline.utils.AttributedStyle;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
/** Utility class that contains all strings for CLI commands and messages. */
public final class CliStrings {
+ /**
+ * A cause caption at the start of a line; a caption inside a message does
not count, and only a
+ * line feed starts a line, since a message may carry other separators in
echoed SQL.
+ */
+ private static final Pattern CAUSED_BY =
+ Pattern.compile("^Caused by: ", Pattern.MULTILINE |
Pattern.UNIX_LINES);
+
+ /**
+ * The end of the exception line of a stack trace segment: its first frame
line, "\tat ..." or
+ * "\t... n more", or the end of the server trace if it has no frames.
+ */
+ private static final Pattern HEADLINE_END =
+ Pattern.compile("\\R(\\tat |\\t\\.\\.\\. \\d+ more|End of
exception on server side>)");
+
+ /** By name, since Calcite is not on the client's runtime classpath. */
+ private static final String CALCITE_PARSE_EXCEPTION =
+ "org.apache.calcite.sql.parser.SqlParseException";
+
private CliStrings() {
// private
}
@@ -313,25 +339,99 @@ public final class CliStrings {
}
public static AttributedString messageError(String message, Throwable t,
boolean isVerbose) {
- while (t.getCause() != null
- && t.getCause().getMessage() != null
- && !t.getCause().getMessage().isEmpty()) {
- t = t.getCause();
- }
-
+ final Throwable reason = findReason(t);
if (isVerbose) {
- return messageError(message, ExceptionUtils.stringifyException(t));
+ return messageError(message,
ExceptionUtils.stringifyException(reason));
+ } else if (reason instanceof RestClientException) {
+ // TODO: Remove this after RestClientException supports to get
RootCause.
+ return messageError(message,
findReasonInStackTrace(reason.getMessage()));
} else {
- if (t instanceof RestClientException) {
- // TODO: Remove this after RestClientException supports to get
RootCause.
- String[] splitExceptions = t.getMessage().split("Caused by: ");
- return messageError(
- message,
- splitExceptions[splitExceptions.length -
1].split("\tat ")[0].trim());
- } else {
- return messageError(message, t.getClass().getName() + ": " +
t.getMessage());
+ return messageError(message, reason.getClass().getName() + ": " +
reason.getMessage());
+ }
+ }
+
+ /**
+ * The root cause, or the outermost exception whose message ends with the
root message and adds
+ * context in front of it, as Calcite does with the position of a
validation error. A planning
+ * rejection, where the plan follows the reason, does not qualify. The
planner's parse exception
+ * is picked by type, since it echoes the line inside the Calcite text.
The search starts at a
+ * {@link RestClientException}, whose message is the whole server trace.
+ */
+ private static Throwable findReason(Throwable t) {
+ final List<Throwable> chain = new ArrayList<>();
+ final Set<Throwable> seen = Collections.newSetFromMap(new
IdentityHashMap<>());
+ int outermost = 0;
+ for (Throwable current = t;
+ current != null && seen.add(current);
+ current = current.getCause()) {
+ chain.add(current);
+ if (current instanceof RestClientException) {
+ outermost = chain.size() - 1;
+ }
+ final Throwable cause = current.getCause();
+ if (cause == null || cause.getMessage() == null ||
cause.getMessage().isEmpty()) {
+ break;
+ }
+ }
+ final Throwable root = chain.get(chain.size() - 1);
+ for (int i = outermost; i < chain.size() - 1; i++) {
+ final Throwable candidate = chain.get(i);
+ final Throwable cause = chain.get(i + 1);
+ final boolean parseError = isParseError(candidate, cause);
+ final boolean quotesRootCause =
addsContext(candidate.getMessage(), root.getMessage());
+ final boolean repeatsCause =
cause.toString().equals(candidate.getMessage());
+ if (parseError || (quotesRootCause && !repeatsCause)) {
+ return candidate;
}
}
+ return root;
+ }
+
+ /** Applies the rule of {@link #findReason} to a stringified stack trace.
*/
+ private static String findReasonInStackTrace(String stackTrace) {
+ final String[] segments = CAUSED_BY.split(stackTrace);
+ final String root = headline(segments[segments.length - 1]);
+ // The first segment is the response preamble, never a cause worth
showing on its own.
+ for (int i = 1; i < segments.length - 1; i++) {
+ final String candidate = headline(segments[i]);
+ final String cause = headline(segments[i + 1]);
+ final boolean parseError = isParseError(candidate, cause);
+ final boolean quotesRootCause = addsContext(messageOf(candidate),
messageOf(root));
+ final boolean repeatsCause = messageOf(candidate).equals(cause);
+ if (parseError || (quotesRootCause && !repeatsCause)) {
+ return candidate;
+ }
+ }
+ return root;
+ }
+
+ private static boolean isParseError(Throwable candidate, Throwable cause) {
+ return candidate instanceof SqlParserException
+ && cause.getClass().getName().equals(CALCITE_PARSE_EXCEPTION);
+ }
+
+ private static boolean isParseError(String headline, String causeHeadline)
{
+ return headline.startsWith(SqlParserException.class.getName() + ": ")
+ && causeHeadline.startsWith(CALCITE_PARSE_EXCEPTION + ": ");
+ }
+
+ private static boolean addsContext(String outer, String inner) {
+ return outer != null
+ && inner != null
+ && !inner.isEmpty()
+ && outer.length() > inner.length()
+ && outer.endsWith(inner);
+ }
+
+ /** The exception line of a stack trace segment, without its frames. */
+ private static String headline(String segment) {
+ final Matcher end = HEADLINE_END.matcher(segment);
+ return (end.find() ? segment.substring(0, end.start()) :
segment).trim();
+ }
+
+ private static String messageOf(String headline) {
+ final int separator = headline.indexOf(": ");
+ return separator < 0 ? "" : headline.substring(separator + 2);
}
public static AttributedString messageError(String message) {
diff --git
a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java
b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java
index 21ba32454d4..2b4282ba054 100644
---
a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java
+++
b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java
@@ -102,6 +102,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
+import java.util.regex.Pattern;
import java.util.stream.Collectors;
import static
org.apache.flink.table.gateway.rest.handler.session.CloseSessionHandler.CLOSE_MESSAGE;
@@ -110,6 +111,15 @@ import static
org.apache.flink.util.ExceptionUtils.firstOrSuppressed;
/** Executor to connect to {@link SqlGateway} and execute statements. */
public class ExecutorImpl implements Executor {
+ /**
+ * The planner's own class for an incomplete statement, as a headline of
the server trace. The
+ * message text is not searched, since the planner echoes the statement in
it.
+ */
+ private static final Pattern INCOMPLETE_STATEMENT =
+ Pattern.compile(
+ "^(Caused by:
)?org\\.apache\\.flink\\.table\\.api\\.SqlParserEOFException: ",
+ Pattern.MULTILINE | Pattern.UNIX_LINES);
+
private static final Logger LOG =
LoggerFactory.getLogger(ExecutorImpl.class);
private static final long HEARTBEAT_INTERVAL_MILLISECONDS = 60_000L;
@@ -496,7 +506,7 @@ public class ExecutorImpl implements Executor {
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof RestClientException
- && cause.getMessage().contains("Encountered \"<EOF>\"")) {
+ &&
INCOMPLETE_STATEMENT.matcher(cause.getMessage()).find()) {
throw new SqlExecutionException(
"The SQL statement is incomplete.",
new SqlParserEOFException(cause.getMessage(), cause));
diff --git
a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliStringsTest.java
b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliStringsTest.java
new file mode 100644
index 00000000000..094d124c5b8
--- /dev/null
+++
b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliStringsTest.java
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.cli;
+
+import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.table.api.SqlParserEOFException;
+import org.apache.flink.table.api.SqlParserException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+
+import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the exception line that {@link CliStrings#messageError} picks. */
+class CliStringsTest {
+
+ private static final String NON_QUERY = "Non-query expression encountered
in illegal context";
+ private static final String POSITIONED_PARSE_ERROR =
+ "SQL parse failed. "
+ + NON_QUERY
+ + ". From line 1, column 1 to line 1, column 5:\n"
+ + " ELECT 1\n"
+ + " ^^^^^";
+
+ private static final String SERVER_PREAMBLE =
+ "[Internal server error., <Exception on server side:\n";
+ private static final String SERVER_END = "End of exception on server
side>]";
+
+ // The chain a parse error takes from the gateway to the client, as the
server prints it.
+ private static final String PARSE_ERROR_BODY =
+ SERVER_PREAMBLE
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: "
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.rest.handler.statement.FetchResultsHandler.handleRequest(FetchResultsHandler.java:91)\n"
+ + "Caused by:
org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:238)\n"
+ + "\t... 3 more\n"
+ + "Caused by:
org.apache.flink.table.gateway.service.utils.SqlExecutionException: Failed to
execute the operation 1.\n"
+ + "\tat
org.apache.flink.table.gateway.service.operation.OperationManager.processThrowable(OperationManager.java:412)\n"
+ + "Caused by:
org.apache.flink.table.api.SqlParserException: "
+ + POSITIONED_PARSE_ERROR
+ + "\n"
+ + "\tat
org.apache.flink.table.planner.parse.CalciteParser.parse(CalciteParser.java:60)\n"
+ + "Caused by:
org.apache.calcite.sql.parser.SqlParseException: "
+ + NON_QUERY
+ + "\n"
+ + "\tat
org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.convertException(FlinkSqlParserImpl.java:512)\n"
+ + "Caused by: org.apache.calcite.runtime.CalciteException:
"
+ + NON_QUERY
+ + "\n"
+ + "\tat
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native
Method)\n"
+ + "\t... 12 more\n"
+ + SERVER_END;
+
+ /** The echoed line sits inside the Calcite text, so the suffix rule
cannot see it. */
+ @Test
+ void testChainShowsTheParseExceptionAroundTheCalciteOne() {
+ final Throwable chain =
+ new SqlExecutionException(
+ "Failed to execute the operation 1.",
+ new SqlParserException(POSITIONED_PARSE_ERROR,
calciteParseException()));
+
+ assertThat(reason(chain))
+ .isEqualTo(
+ "org.apache.flink.table.api.SqlParserException: " +
POSITIONED_PARSE_ERROR);
+ }
+
+ @Test
+ void testChainShowsAParseExceptionWithoutCauseAsTheRootCause() {
+ final Throwable chain =
+ new SqlExecutionException(
+ "Failed to execute the operation 1.",
+ new SqlParserException("SQL parse failed. boom"));
+
+ assertThat(reason(chain))
+ .isEqualTo("org.apache.flink.table.api.SqlParserException: SQL
parse failed. boom");
+ }
+
+ @Test
+ void testChainShowsTheOutermostExceptionThatQuotesTheRootCause() {
+ final String notFound = "Column 'x' not found in any table";
+ final String positioned = "From line 1, column 8 to line 1, column 9:
" + notFound;
+ final Throwable chain =
+ new SqlExecutionException(
+ "Failed to execute the operation 1.",
+ new ValidationException(
+ "SQL validation failed. " + positioned,
+ new RuntimeException(positioned, new
RuntimeException(notFound))));
+
+ assertThat(reason(chain))
+ .isEqualTo(
+ "org.apache.flink.table.api.ValidationException: SQL
validation failed. "
+ + positioned);
+ }
+
+ @Test
+ void testChainShowsTheStatementThatFailed() {
+ final Throwable chain =
+ new SqlExecutionException(
+ "Failed to execute the operation 1.",
+ new ValidationException(
+ "Could not execute LOAD MODULE core. A module
with name 'core' already exists",
+ new ValidationException(
+ "A module with name 'core' already
exists")));
+
+ assertThat(reason(chain))
+ .isEqualTo(
+ "org.apache.flink.table.api.ValidationException: Could
not execute LOAD MODULE"
+ + " core. A module with name 'core' already
exists");
+ }
+
+ @Test
+ void testChainShowsTheRootCauseWhenNothingQuotesIt() {
+ final Throwable chain =
+ new SqlExecutionException(
+ "Failed to execute the operation 1.", new
RuntimeException("boom"));
+
+ assertThat(reason(chain)).isEqualTo("java.lang.RuntimeException:
boom");
+ }
+
+ @Test
+ void testChainIgnoresAWrapperWithTheSameMessage() {
+ final Throwable chain = new ValidationException("boom", new
RuntimeException("boom"));
+
+ assertThat(reason(chain)).isEqualTo("java.lang.RuntimeException:
boom");
+ }
+
+ @Test
+ void testChainIgnoresAWrapperThatOnlyAddsItsClassName() {
+ final Throwable chain = new RuntimeException(new
IllegalStateException("boom"));
+
+ assertThat(reason(chain)).isEqualTo("java.lang.IllegalStateException:
boom");
+ }
+
+ @Test
+ void testChainStopsAtAnEmptyMessage() {
+ final Throwable chain = new RuntimeException("outer", new
IllegalStateException(""));
+
+ assertThat(reason(chain)).isEqualTo("java.lang.RuntimeException:
outer");
+ }
+
+ @Test
+ void testChainKeepsTheIncompleteStatementLine() {
+ final String body =
+ SERVER_PREAMBLE
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:238)\n"
+ + "Caused by:
org.apache.flink.sql.parser.impl.ParseException: Encountered \"<EOF>\" at line
1, column 13.\n"
+ + "Was expecting one of:\n"
+ + " \"AS\" ...\n"
+ + " \n"
+ + "\tat
org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.generateParseException(FlinkSqlParserImpl.java:1)\n"
+ + SERVER_END;
+ // ExecutorImpl repeats the whole body in the wrapper it builds for an
incomplete statement.
+ final Throwable chain =
+ new SqlExecutionException(
+ "The SQL statement is incomplete.",
+ new SqlParserEOFException(
+ body,
+ new RestClientException(
+ body,
HttpResponseStatus.INTERNAL_SERVER_ERROR)));
+
+ assertThat(reason(chain))
+ .isEqualTo(
+ "org.apache.flink.sql.parser.impl.ParseException:
Encountered \"<EOF>\" at"
+ + " line 1, column 13.\nWas expecting one
of:\n \"AS\" ...");
+ }
+
+ @Test
+ void testCyclicCausesTerminate() {
+ final RuntimeException inner = new RuntimeException("b");
+ final RuntimeException outer = new RuntimeException("a", inner);
+ inner.initCause(outer);
+
+ assertThat(reason(outer)).isEqualTo("java.lang.RuntimeException: b");
+ }
+
+ @Test
+ void testRestBodyShowsTheParseExceptionAroundTheCalciteOne() {
+ assertThat(reason(restFailure(PARSE_ERROR_BODY)))
+ .isEqualTo(
+ "org.apache.flink.table.api.SqlParserException: " +
POSITIONED_PARSE_ERROR);
+ }
+
+ @Test
+ void testRestBodyShowsTheRootCauseWhenNothingQuotesIt() {
+ final String body =
+ "[org.apache.flink.runtime.rest.handler.RestHandlerException:
Failed to cancelOperation.\n"
+ + "\tat
org.apache.flink.table.gateway.rest.handler.operation.AbstractOperationHandler.handleRequest(AbstractOperationHandler.java:79)\n"
+ + "Caused by: java.lang.RuntimeException: boom\n"
+ + "\tat
org.apache.flink.table.gateway.service.operation.OperationManager.cancelOperation(OperationManager.java:1)\n"
+ + "]";
+
+
assertThat(reason(restFailure(body))).isEqualTo("java.lang.RuntimeException:
boom");
+ }
+
+ @Test
+ void testRestBodyWithoutCausesIsShownAsIs() {
+ final String body = "[The endpoint has not started yet.]";
+
+ assertThat(reason(restFailure(body))).isEqualTo(body);
+ }
+
+ @Test
+ void testRestBodyCutsTheRootCauseAtItsFrames() {
+ final String body =
+ SERVER_PREAMBLE
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:238)\n"
+ + "Caused by: java.lang.RuntimeException: boom\n"
+ + "\t... 3 more\n"
+ + SERVER_END;
+
+
assertThat(reason(restFailure(body))).isEqualTo("java.lang.RuntimeException:
boom");
+ }
+
+ @Test
+ void testRestBodyCutsARootCauseWithoutFramesAtTheServerEnd() {
+ final String body =
+ SERVER_PREAMBLE
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:238)\n"
+ + "Caused by: java.lang.RuntimeException: boom\n"
+ + "\n"
+ + SERVER_END;
+
+
assertThat(reason(restFailure(body))).isEqualTo("java.lang.RuntimeException:
boom");
+ }
+
+ @Test
+ void testRestBodyIgnoresAWrapperThatOnlyAddsItsClassName() {
+ final String body =
+ SERVER_PREAMBLE
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:238)\n"
+ + "Caused by:
java.util.concurrent.CompletionException:
org.apache.flink.table.api.ValidationException: boom\n"
+ + "\tat
java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315)\n"
+ + "Caused by:
org.apache.flink.table.api.ValidationException: boom\n"
+ + "\t... 3 more\n"
+ + SERVER_END;
+
+ assertThat(reason(restFailure(body)))
+ .isEqualTo("org.apache.flink.table.api.ValidationException:
boom");
+ }
+
+ @Test
+ void testRestBodyKeepsATabIndentedLineOfAMessage() {
+ final String body =
+ SERVER_PREAMBLE
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:238)\n"
+ + "Caused by: java.lang.RuntimeException: boom\n"
+ + "\tdetail\n"
+ + "\tat
org.apache.flink.table.gateway.service.operation.OperationManager.cancelOperation(OperationManager.java:1)\n"
+ + SERVER_END;
+
+ assertThat(reason(restFailure(body)))
+ .isEqualTo("java.lang.RuntimeException: boom\n\tdetail");
+ }
+
+ @Test
+ void testRestBodyIgnoresACauseCaptionInsideAMessage() {
+ final String notFound = "Column 'Caused by: x' not found in any table";
+ final String positioned = "From line 1, column 8 to line 1, column 21:
" + notFound;
+ final String body =
+ SERVER_PREAMBLE
+ +
"org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to
fetchResults.\n"
+ + "\tat
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:238)\n"
+ + "Caused by:
org.apache.flink.table.api.ValidationException: SQL validation failed. "
+ + positioned
+ + "\n"
+ + "\tat
org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:205)\n"
+ + "Caused by:
org.apache.calcite.runtime.CalciteContextException: "
+ + positioned
+ + "\n"
+ + "\tat
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native
Method)\n"
+ + "Caused by:
org.apache.calcite.sql.validate.SqlValidatorException: "
+ + notFound
+ + "\n"
+ + "\tat
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native
Method)\n"
+ + SERVER_END;
+
+ assertThat(reason(restFailure(body)))
+ .isEqualTo(
+ "org.apache.flink.table.api.ValidationException: SQL
validation failed. "
+ + positioned);
+ }
+
+ @Test
+ void testVerboseOutputStartsAtTheChosenException() {
+ final Throwable chain =
+ new SqlExecutionException(
+ "Failed to execute the operation 1.",
+ new SqlParserException(POSITIONED_PARSE_ERROR,
calciteParseException()));
+
+ final String output =
+ CliStrings.messageError("Could not execute SQL statement.",
chain, true).toString();
+
+ assertThat(output)
+ .contains(
+ "org.apache.flink.table.api.SqlParserException: " +
POSITIONED_PARSE_ERROR)
+ .contains(
+ "Caused by:
org.apache.calcite.sql.parser.SqlParseException: " + NON_QUERY)
+ .doesNotContain("Failed to execute the operation 1.");
+ }
+
+ private static SqlParseException calciteParseException() {
+ return new SqlParseException(NON_QUERY, null, null, null, null);
+ }
+
+ private static Throwable restFailure(String body) {
+ return new SqlExecutionException(
+ "Failed to get response for the operation 1.",
+ new RestClientException(body,
HttpResponseStatus.INTERNAL_SERVER_ERROR));
+ }
+
+ private static String reason(Throwable t) {
+ final String output =
+ CliStrings.messageError("Could not execute SQL statement.", t,
false).toString();
+ final String header = "[ERROR] Could not execute SQL statement.
Reason:\n";
+ assertThat(output).startsWith(header);
+ return output.substring(header.length());
+ }
+}
diff --git
a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java
b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java
index ea304bc97bd..c9546f0fd81 100644
---
a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java
+++
b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
+import org.apache.flink.table.api.SqlParserEOFException;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
@@ -68,6 +69,7 @@ import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.test.util.TestBaseUtils;
import org.apache.flink.test.util.TestUtils;
import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.UserClassLoaderJarTestUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
@@ -218,6 +220,34 @@ class ExecutorImplITCase {
}
}
+ @Test
+ void testEchoedStatementTextIsNotTakenForAnIncompleteStatement() {
+ try (Executor executor = createRestServiceExecutor()) {
+ initSession(executor, Collections.emptyMap());
+ assertThatThrownBy(
+ () ->
+ fetchAll(
+ executor,
+ "SELECT 1 UNION foo_bar --
Encountered \"<EOF>\""))
+ .satisfies(
+ e ->
+ assertThat(
+
ExceptionUtils.findThrowable(
+ e,
SqlParserEOFException.class))
+ .isEmpty())
+ .satisfies(
+ anyCauseMatches("Non-query expression encountered
in illegal context"));
+ }
+ }
+
+ private static void fetchAll(Executor executor, String statement) {
+ try (StatementResult result = executor.executeStatement(statement)) {
+ while (result.hasNext()) {
+ result.next();
+ }
+ }
+ }
+
@Test
void testStreamQueryExecutionChangelog() throws Exception {
final URL url =
getClass().getClassLoader().getResource("test-data.csv");
diff --git
a/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
b/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
index 50f61a566de..2af27ae68e3 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
@@ -31,7 +31,9 @@ org.apache.flink.table.catalog.exceptions.CatalogException: A
catalog with name
create catalog invalid.cat with ('type'='generic_in_memory');
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Encountered "." at line 1,
column 23.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered
"." at line 1, column 23:
+ create catalog invalid.cat with ('type'='generic_in_memory');
+ ^
Was expecting one of:
<EOF>
"WITH" ...
@@ -523,7 +525,7 @@ show databases from c0 ilike 'db%';
show databases in c0.t;
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Show databases from/in
identifier [ c0.t ] format error, catalog must be a single part identifier.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Show
databases from/in identifier [ c0.t ] format error, catalog must be a single
part identifier.
!error
drop catalog `c0`;
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/module.q
b/flink-table/flink-sql-client/src/test/resources/sql/module.q
index 9f82c9281ff..d4bfc9e5c8a 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/module.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/module.q
@@ -46,19 +46,21 @@ SHOW FULL MODULES;
# load core module twice
LOAD MODULE core;
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.table.api.ValidationException: A module with name 'core'
already exists
+org.apache.flink.table.api.ValidationException: Could not execute LOAD MODULE
`core`. A module with name 'core' already exists
!error
# use hive built-in function without loading hive module
SELECT SUBSTRING_INDEX('www.apache.org', '.', 2) FROM (VALUES (1, 'Hello
World')) AS T(id, str);
[ERROR] Could not execute SQL statement. Reason:
-org.apache.calcite.sql.validate.SqlValidatorException: No match found for
function signature SUBSTRING_INDEX(<CHARACTER>, <CHARACTER>, <NUMERIC>)
+org.apache.flink.table.api.ValidationException: SQL validation failed. From
line 1, column 8 to line 1, column 48: No match found for function signature
SUBSTRING_INDEX(<CHARACTER>, <CHARACTER>, <NUMERIC>)
!error
# load dummy module with module name as string literal
LOAD MODULE 'dummy';
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Encountered "\'dummy\'" at
line 1, column 13.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered
"\'dummy\'" at line 1, column 13:
+ LOAD MODULE 'dummy';
+ ^^^^^^^
Was expecting one of:
<BRACKET_QUOTED_IDENTIFIER> ...
<QUOTED_IDENTIFIER> ...
@@ -83,7 +85,7 @@ dummy
# load dummy module with specifying type
LOAD MODULE mydummy WITH ('type' = 'dummy');
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.table.api.ValidationException: Option 'type' = 'dummy' is not
supported since module name is used to find module
+org.apache.flink.table.api.ValidationException: Could not execute LOAD MODULE
`mydummy` WITH ('type' = 'dummy'). Option 'type' = 'dummy' is not supported
since module name is used to find module
!error
LOAD MODULE dummy;
@@ -119,7 +121,7 @@ SHOW FULL MODULES;
# use duplicate modules
USE MODULES dummy, core, dummy;
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.table.api.ValidationException: Module 'dummy' appears more
than once
+org.apache.flink.table.api.ValidationException: Could not execute USE MODULES:
[dummy, core, dummy]. Module 'dummy' appears more than once
!error
# change module resolution order
@@ -195,5 +197,5 @@ SHOW FULL MODULES;
# unload core module twice
UNLOAD MODULE core;
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.table.api.ValidationException: No module with name 'core'
exists
+org.apache.flink.table.api.ValidationException: Could not execute UNLOAD
MODULE core. No module with name 'core' exists
!error
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/select.q
b/flink-table/flink-sql-client/src/test/resources/sql/select.q
index 81657060f3a..b33c215bd47 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/select.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/select.q
@@ -221,6 +221,13 @@ SET 'table.display.max-column-width' = '30';
[INFO] Execute statement succeeded.
!info
+# a planning rejection prints the reason, not the plan that follows it
+
+SELECT * FROM (VALUES (1, 2), (3, 4)) AS t(a, b) ORDER BY a;
+[ERROR] Could not execute SQL statement. Reason:
+org.apache.flink.table.api.TableException: Streaming ORDER BY requires the
primary sort key to be a time attribute in ascending order, but 'a' is INT NOT
NULL. A time attribute is an event-time column (a TIMESTAMP or TIMESTAMP_LTZ
column with a WATERMARK) or a processing-time column. Otherwise use LIMIT for
Top-N, sort within a window, or run in batch mode.
+!error
+
# ==========================================================================
# test batch query
# ==========================================================================
@@ -367,3 +374,28 @@ SELECT --;
+--------+
1 row in set
!ok
+
+# parse errors name the position of the token the parser stopped at
+
+ELECT 1;
+[ERROR] Could not execute SQL statement. Reason:
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Non-query
expression encountered in illegal context. From line 1, column 1 to line 1,
column 5:
+ ELECT 1;
+ ^^^^^
+!error
+
+/*
+comment
+*/ ELECT 1;
+[ERROR] Could not execute SQL statement. Reason:
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Non-query
expression encountered in illegal context. From line 3, column 5 to line 3,
column 9:
+ */ ELECT 1;
+ ^^^^^
+!error
+
+SELECT * FROM (t);
+[ERROR] Could not execute SQL statement. Reason:
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Expected
query or join. At line 1, column 16:
+ SELECT * FROM (t);
+ ^
+!error
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 2987d28fc9c..7419a55b74e 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
@@ -123,7 +123,9 @@ CREATE TABLE hive_table2 (
'streaming-source.enable' = 'true'
);
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Encountered "STRING" at line
10, column 27.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered
"STRING" at line 10, column 27:
+ ) PARTITIONED BY (pt_year STRING, pt_month STRING, pt_day STRING)
TBLPROP...
+ ^^^^^^
Was expecting one of:
")" ...
"," ...
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/table.q
b/flink-table/flink-sql-client/src/test/resources/sql/table.q
index b5caeafb5cc..e7a7a250562 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/table.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/table.q
@@ -21,7 +21,7 @@
create table tbl(a int, b as invalid_function());
[ERROR] Could not execute SQL statement. Reason:
-org.apache.calcite.sql.validate.SqlValidatorException: No match found for
function signature invalid_function()
+org.apache.calcite.runtime.CalciteContextException: From line 1, column 30 to
line 1, column 47: No match found for function signature invalid_function()
!error
drop table non_exist;
diff --git
a/flink-table/flink-sql-client/src/test/resources/sql_multi/statement_set.q
b/flink-table/flink-sql-client/src/test/resources/sql_multi/statement_set.q
index 0f223fb6172..8f1f6c04f90 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql_multi/statement_set.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql_multi/statement_set.q
@@ -178,7 +178,9 @@ EXPLAIN STATEMENT SET BEGIN
END;
!output
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Encountered "END" at line 2,
column 1.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered
"END" at line 2, column 1:
+ END;
+ ^^^
Was expecting one of:
"INSERT" ...
"UPSERT" ...
@@ -188,7 +190,9 @@ EXECUTE STATEMENT SET BEGIN
END;
!output
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Encountered "END" at line 2,
column 1.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered
"END" at line 2, column 1:
+ END;
+ ^^^
Was expecting one of:
"INSERT" ...
"UPSERT" ...
@@ -306,7 +310,9 @@ EXPLAIN STATEMENT SET BEGIN
END;
!output
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Encountered "END" at line 2,
column 1.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered
"END" at line 2, column 1:
+ END;
+ ^^^
Was expecting one of:
"INSERT" ...
"UPSERT" ...
@@ -316,7 +322,9 @@ EXECUTE STATEMENT SET BEGIN
END;
!output
[ERROR] Could not execute SQL statement. Reason:
-org.apache.flink.sql.parser.impl.ParseException: Encountered "END" at line 2,
column 1.
+org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered
"END" at line 2, column 1:
+ END;
+ ^^^
Was expecting one of:
"INSERT" ...
"UPSERT" ...
