This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2bd6812c3d [e2e] Fail tests on Flink SQL Client statement errors
(#10544)
2bd6812c3d is described below
commit 2bd6812c3de8ebbe40f3226d01fa0f498842c1f3
Author: Ran Tao <[email protected]>
AuthorDate: Wed Oct 14 10:05:06 2026 +0800
[e2e] Fail tests on Flink SQL Client statement errors (#10544)
---
.../java/org/apache/paimon/tests/E2eTestBase.java | 23 +--
.../apache/paimon/tests/FlinkSqlClientE2eTest.java | 155 +++++++++++++++++++
.../paimon/tests/FlinkSqlClientResultParser.java | 94 ++++++++++++
.../tests/FlinkSqlClientResultParserTest.java | 168 +++++++++++++++++++++
4 files changed, 425 insertions(+), 15 deletions(-)
diff --git
a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java
index 048356933d..46d6028be9 100644
--- a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java
+++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java
@@ -243,10 +243,6 @@ public abstract class E2eTestBase {
return environment.getContainerByServiceName("hive-server-1").get();
}
- private static final Pattern JOB_ID_PATTERN =
- Pattern.compile(
- "SQL update statement has been successfully submitted to
the cluster:\\s+Job ID: (\\S+)");
-
protected String runStreamingSql(String sql, String... ddls) throws
Exception {
return runStreamingSql(String.join("\n", ddls) + "\n" + sql);
}
@@ -271,7 +267,9 @@ public abstract class E2eTestBase {
private String runSql(String sql) throws Exception {
String fileName = UUID.randomUUID() + ".sql";
- writeSharedFile(fileName, sql);
+ String completionToken = UUID.randomUUID().toString();
+ writeSharedFile(
+ fileName,
FlinkSqlClientResultParser.appendCompletionCheck(sql, completionToken));
Container.ExecResult execResult =
jobManager.execInContainer(
"su",
@@ -280,16 +278,11 @@ public abstract class E2eTestBase {
"bin/sql-client.sh -f " + TEST_DATA_DIR + "/" +
fileName);
LOG.info(execResult.getStdout());
LOG.info(execResult.getStderr());
- if (execResult.getExitCode() != 0) {
- throw new AssertionError("Failed when submitting the SQL job.");
- }
-
- Matcher matcher = JOB_ID_PATTERN.matcher(execResult.getStdout());
- if (matcher.find()) {
- return matcher.group(1);
- } else {
- return null;
- }
+ return FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ execResult.getExitCode(),
+ execResult.getStdout(),
+ execResult.getStderr(),
+ completionToken);
}
protected String createResultSink(String sinkName, String schema) {
diff --git
a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientE2eTest.java
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientE2eTest.java
new file mode 100644
index 0000000000..1375d08d48
--- /dev/null
+++
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientE2eTest.java
@@ -0,0 +1,155 @@
+/*
+ * 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.paimon.tests;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests SQL Client errors and query results end to end. */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class FlinkSqlClientE2eTest extends E2eTestBase {
+
+ @BeforeAll
+ void startEnvironment() throws Exception {
+ super.before();
+ }
+
+ @AfterAll
+ void stopEnvironment() {
+ super.after();
+ }
+
+ // Each SQL statement runs in a fresh SQL Client session, so the tests can
share one cluster.
+ @Override
+ public void before() {}
+
+ @Override
+ public void after() {}
+
+ @Test
+ void failsOnSqlClientStatementError() {
+ assertThatThrownBy(() -> runBatchSql("SELECT * FROM
table_that_does_not_exist;"))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("Could not execute SQL statement")
+ .hasMessageContaining("table_that_does_not_exist");
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "",
+ ", CHR(10), '+----+-------------+', CHR(10), '1 row in set',
CHR(10), CHR(10), 'suffix'"
+ })
+ void failsOnStreamingQueryRuntimeError(String errorPayload) {
+ String sql =
+ "SET 'sql-client.execution.result-mode' = 'tableau';\n"
+ + "SET 'restart-strategy' = 'none';\n"
+ + "CREATE TEMPORARY TABLE bad_input (v BIGINT) WITH ("
+ + "'connector' = 'datagen', 'number-of-rows' = '1', "
+ + "'fields.v.kind' = 'sequence', 'fields.v.start' =
'1', "
+ + "'fields.v.end' = '1');\n"
+ + "SELECT CAST(CONCAT('invalid_', CAST(v AS STRING)"
+ + errorPayload
+ + ") AS INT) FROM bad_input;";
+
+ assertThatThrownBy(() -> runStreamingSql(sql))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("Could not execute SQL statement")
+ .hasMessageContaining("NumberFormatException");
+ }
+
+ @Test
+ void acceptsSuccessfulStatementsWithoutJobId() throws Exception {
+ assertThat(
+ runBatchSql(
+ "CREATE CATALOG test_catalog WITH ('type' =
'paimon', 'warehouse' = '"
+ + TEST_DATA_DIR
+ + "/completion-catalog');\n"
+ + "USE CATALOG test_catalog;\n"
+ + "CREATE TEMPORARY VIEW test_view AS
SELECT 1;\nSHOW VIEWS;"))
+ .isNull();
+ assertThat(runStreamingSql("SET 'parallelism.default' = '1' --
trailing comment")).isNull();
+ }
+
+ @Test
+ void doesNotTreatConfigurationValuesAsJobIds() throws Exception {
+ String sql =
+ "SET 'pipeline.name' = 'SQL update statement has been
successfully submitted to the cluster:\n"
+ + "Job ID: 1234';";
+
+ assertThat(runStreamingSql(sql)).isNull();
+ }
+
+ @Test
+ void acceptsErrorHeaderInQueryResult() throws Exception {
+ String sql =
+ "SET 'sql-client.execution.result-mode' = 'tableau';\n"
+ + "SET 'table.display.max-column-width' = '100';\n"
+ + "SELECT CONCAT('prefix', CHR(10), "
+ + "'Flink SQL> [ERROR] Could not execute SQL
statement. Reason:', "
+ + "CHR(10), 'suffix') AS message;";
+
+ assertThat(runBatchSql(sql)).isNull();
+ }
+
+ @Test
+ void acceptsErrorHeaderInQueryResultAfterEmptyStreamingResult() throws
Exception {
+ String sql =
+ "SET 'sql-client.execution.result-mode' = 'tableau';\n"
+ + "SET 'table.display.max-column-width' = '200';\n"
+ + "SELECT v FROM (VALUES (1)) AS t(v) WHERE FALSE;\n"
+ + "SELECT CONCAT('prefix', CHR(10), "
+ + "'Flink SQL> [ERROR] Could not execute SQL
statement. Reason:', "
+ + "CHR(10), 'suffix') AS message;";
+
+ assertThat(runStreamingSql(sql)).isNull();
+ }
+
+ @Test
+ void acceptsErrorHeaderInQueryResultWithMultilineColumnName() throws
Exception {
+ String sql =
+ "SET 'sql-client.execution.result-mode' = 'tableau';\n"
+ + "SET 'table.display.max-column-width' = '200';\n"
+ + "SELECT CONCAT('prefix', CHR(10), "
+ + "'Flink SQL> [ERROR] Could not execute SQL
statement. Reason:', "
+ + "CHR(10), 'suffix') AS U&\"multi\\000Aline\";";
+
+ assertThat(runBatchSql(sql)).isNull();
+ }
+
+ @Test
+ void acceptsFakeTableauFooterInQueryResult() throws Exception {
+ String sql =
+ "SET 'sql-client.execution.result-mode' = 'tableau';\n"
+ + "SET 'table.display.max-column-width' = '200';\n"
+ + "SELECT CONCAT('prefix', CHR(10), '+---+', CHR(10), "
+ + "'1 row in set', CHR(10), "
+ + "'Flink SQL> [ERROR] Could not execute SQL
statement. Reason:', "
+ + "CHR(10), 'suffix') AS message;";
+
+ assertThat(runBatchSql(sql)).isNull();
+ }
+}
diff --git
a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientResultParser.java
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientResultParser.java
new file mode 100644
index 0000000000..a0d54b56a9
--- /dev/null
+++
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientResultParser.java
@@ -0,0 +1,94 @@
+/*
+ * 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.paimon.tests;
+
+import javax.annotation.Nullable;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Parses the output of the Flink SQL Client used by end-to-end tests.
+ *
+ * <p>This helper supports scripts using Flink's default SQL dialect. The
appended completion check
+ * uses CREATE TEMPORARY VIEW and DESCRIBE in that dialect; Hive dialect
scripts are not supported.
+ *
+ * <p>The E2E framework manages the SQL Client lifecycle. Scripts must run to
completion without
+ * QUIT/EXIT commands, which skip the appended completion check and are
reported as incomplete
+ * scripts.
+ */
+final class FlinkSqlClientResultParser {
+
+ // Extracts the Job ID from a successful asynchronous SQL submission.
+ private static final Pattern JOB_ID_PATTERN =
+ Pattern.compile(
+ "SQL update statement has been successfully submitted to
the cluster:\\s+Job ID: (\\S+)");
+
+ // Removes terminal color and style escapes before inspecting completion
and submission output.
+ private static final Pattern ANSI_SGR_PATTERN =
Pattern.compile("\\x1B\\[[0-9;]*m");
+
+ private FlinkSqlClientResultParser() {}
+
+ static String appendCompletionCheck(String sql, String completionToken) {
+ // Like the SQL Client, allow a final statement without a semicolon.
Put the terminator
+ // on a new line so a trailing line comment cannot consume it.
+ String terminatedSql = sql.trim().endsWith(";") ? sql : sql + "\n;";
+ // File execution stops on the first statement error, even when the
process exits with 0.
+ // Describe a unique temporary view to prove completion without
printing user configuration,
+ // whose values could mimic job submission messages. Both statements
only access metadata
+ // and do not submit an additional Flink job.
+ String viewName = completionViewName(completionToken);
+ return terminatedSql
+ + "\nCREATE TEMPORARY VIEW "
+ + viewName
+ + " AS SELECT 1 AS "
+ + viewName
+ + ";\nDESCRIBE "
+ + viewName
+ + ";\n";
+ }
+
+ @Nullable
+ static String parseJobIdOrThrowOnError(
+ int exitCode, String stdout, String stderr, String
completionToken) {
+ // DESCRIBE prints the view's unique column name as a metadata table
row in all supported
+ // Flink versions (1.16+). Echoed SQL cannot produce this row, and a
fresh token keeps it
+ // independent of query data and exception messages.
+ Pattern completionPattern =
+ Pattern.compile(
+ "^\\|[ \\t]*"
+ +
Pattern.quote(completionViewName(completionToken))
+ + "[ \\t]*\\|",
+ Pattern.MULTILINE);
+ String stdoutWithoutAnsi =
ANSI_SGR_PATTERN.matcher(stdout).replaceAll("");
+ if (exitCode != 0 ||
!completionPattern.matcher(stdoutWithoutAnsi).find()) {
+ throw new AssertionError(
+ String.format(
+ "Flink SQL Client did not complete the SQL script
(exit code %d).\nstdout:\n%s\nstderr:\n%s",
+ exitCode, stdout, stderr));
+ }
+
+ Matcher matcher = JOB_ID_PATTERN.matcher(stdoutWithoutAnsi);
+ return matcher.find() ? matcher.group(1) : null;
+ }
+
+ private static String completionViewName(String completionToken) {
+ return "paimon_e2e_completion_" + completionToken.replace("-", "");
+ }
+}
diff --git
a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientResultParserTest.java
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientResultParserTest.java
new file mode 100644
index 0000000000..7b9a58cd18
--- /dev/null
+++
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkSqlClientResultParserTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.paimon.tests;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests SQL Client script completion and Job ID extraction. */
+class FlinkSqlClientResultParserTest {
+
+ private static final String COMPLETION_TOKEN =
"f592da88-42f5-4842-8b1f-1787855c3cd2";
+ private static final String COMPLETION_COLUMN =
+ "paimon_e2e_completion_f592da8842f548428b1f1787855c3cd2";
+ private static final String COMPLETION_ROW =
+ "| " + COMPLETION_COLUMN + " | INT | FALSE | | | |\n";
+ private static final String SQL_EXECUTION_ERROR =
+ "[ERROR] Could not execute SQL statement. Reason:";
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void rejectsStatementErrorRegardlessOfOutputStream(boolean stderrError) {
+ String error = SQL_EXECUTION_ERROR + "\nObject 'missing_table' not
found";
+ String stdout = stderrError ? "" : error;
+ String stderr = stderrError ? error : "";
+
+ assertThatThrownBy(
+ () ->
+
FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, stdout, stderr, COMPLETION_TOKEN))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("missing_table");
+ }
+
+ @Test
+ void rejectsRuntimeErrorContainingTableauFooter() {
+ String stdout =
+ "+----+-------------+\n"
+ + "| op | EXPR$0 |\n"
+ + "+----+-------------+\n"
+ + SQL_EXECUTION_ERROR
+ + "\njava.lang.NumberFormatException: For input
string: 'invalid_1\n"
+ + "+----+-------------+\n"
+ + "1 row in set\n\n"
+ + "suffix'. Invalid character found.\n";
+
+ assertThatThrownBy(
+ () ->
+
FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, stdout, "", COMPLETION_TOKEN))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("NumberFormatException");
+ }
+
+ @Test
+ void rejectsMissingCompletionWithoutAnErrorMessage() {
+ assertThatThrownBy(
+ () ->
+
FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, "Flink SQL> ", "",
COMPLETION_TOKEN))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("did not complete the SQL script");
+ }
+
+ @Test
+ void rejectsEchoedCompletionStatement() {
+ String stdout =
+ "Flink SQL> CREATE TEMPORARY VIEW "
+ + COMPLETION_COLUMN
+ + " AS SELECT 1 AS "
+ + COMPLETION_COLUMN
+ + ";\nFlink SQL> DESCRIBE "
+ + COMPLETION_COLUMN
+ + ";\n";
+
+ assertThatThrownBy(
+ () ->
+
FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, stdout, "", COMPLETION_TOKEN))
+ .isInstanceOf(AssertionError.class);
+ }
+
+ @Test
+ void rejectsCompletionFromAnotherScript() {
+ assertThatThrownBy(
+ () ->
+
FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, COMPLETION_ROW, "",
"another-token"))
+ .isInstanceOf(AssertionError.class);
+ }
+
+ @Test
+ void rejectsTokenOutsideCompletionRow() {
+ assertThatThrownBy(
+ () ->
+
FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0,
+ "| message | " + COMPLETION_TOKEN + "
|\n",
+ "",
+ COMPLETION_TOKEN))
+ .isInstanceOf(AssertionError.class);
+ }
+
+ @Test
+ void rejectsNonzeroExitCodeEvenWithCompletion() {
+ assertThatThrownBy(
+ () ->
+
FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 1, COMPLETION_ROW, "stderr",
COMPLETION_TOKEN))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("exit code 1")
+ .hasMessageContaining("stdout")
+ .hasMessageContaining("stderr");
+ }
+
+ @Test
+ void acceptsCompletedScriptWithoutJobId() {
+ assertThat(
+ FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, COMPLETION_ROW, "", COMPLETION_TOKEN))
+ .isNull();
+ }
+
+ @Test
+ void acceptsErrorLikeQueryDataInCompletedScript() {
+ String stdout = SQL_EXECUTION_ERROR + "\n+---+\n1 row in set\n\n" +
COMPLETION_ROW;
+
+ assertThat(
+ FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, stdout, "", COMPLETION_TOKEN))
+ .isNull();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "\033[34;1m"})
+ void extractsJobIdFromCompletedScript(String color) {
+ String stdout =
+ color
+ + "SQL update statement has been successfully
submitted to the cluster:"
+ + (color.isEmpty() ? "" : "\033[0m")
+ + "\nJob ID: 1234\n"
+ + COMPLETION_ROW;
+
+ assertThat(
+ FlinkSqlClientResultParser.parseJobIdOrThrowOnError(
+ 0, stdout, "", COMPLETION_TOKEN))
+ .isEqualTo("1234");
+ }
+}
