This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-paimon-webui.git
The following commit(s) were added to refs/heads/main by this push:
new 663965a [Feature] Introduce Flink SQL gateway baseline implementation
of job submission (#153)
663965a is described below
commit 663965abca32d59d75d912ebbe5f8f3b2aa8ce75
Author: s7monk <[email protected]>
AuthorDate: Thu Feb 1 16:54:30 2024 +0800
[Feature] Introduce Flink SQL gateway baseline implementation of job
submission (#153)
---
.../web/engine/flink/common/executor/Executor.java | 3 +-
.../flink/common/executor/ExecutorFactory.java | 31 +++
.../paimon-web-engine-flink-sql-gateway}/pom.xml | 18 +-
.../flink/sql/gateway/client/SqlGatewayClient.java | 245 +++++++++++++++++++++
.../gateway/executor/FlinkSqlGatewayExecutor.java | 163 ++++++++++++++
.../executor/FlinkSqlGatewayExecutorFactory.java | 38 ++++
.../flink/sql/gateway/model/SessionEntity.java | 37 ++++
.../flink/sql/gateway/utils/CollectResultUtil.java | 63 ++++++
.../gateway/utils/FlinkSqlStatementSetBuilder.java | 61 +++++
.../sql/gateway/utils/SqlGateWayRestClient.java | 147 +++++++++++++
.../web/engine/flink/sql/gataway/TestBase.java | 201 +++++++++++++++++
.../sql/gataway/client/SqlGatewayClientTest.java | 109 +++++++++
.../executor/FlinkSqlGatewayExecutorTest.java | 116 ++++++++++
.../sql/gataway/executor/StatementsConstant.java | 196 +++++++++++++++++
paimon-web-engine/paimon-web-engine-flink/pom.xml | 1 +
paimon-web-engine/pom.xml | 8 +
16 files changed, 1427 insertions(+), 10 deletions(-)
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-common/src/main/java/org/apache/paimon/web/engine/flink/common/executor/Executor.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-common/src/main/java/org/apache/paimon/web/engine/flink/common/executor/Executor.java
index 64218a1..60a754f 100644
---
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-common/src/main/java/org/apache/paimon/web/engine/flink/common/executor/Executor.java
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-common/src/main/java/org/apache/paimon/web/engine/flink/common/executor/Executor.java
@@ -50,8 +50,7 @@ public interface Executor {
*
* @param jobId The unique identifier of the job to stop.
* @param withSavepoint If true, the job will create a savepoint before
stopping.
- * @return true if the job was successfully stopped, false otherwise.
* @throws Exception if the job cannot be stopped or savepoint cannot be
created.
*/
- boolean stop(String jobId, boolean withSavepoint) throws Exception;
+ void stop(String jobId, boolean withSavepoint) throws Exception;
}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-common/src/main/java/org/apache/paimon/web/engine/flink/common/executor/ExecutorFactory.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-common/src/main/java/org/apache/paimon/web/engine/flink/common/executor/ExecutorFactory.java
new file mode 100644
index 0000000..5d87fe0
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-common/src/main/java/org/apache/paimon/web/engine/flink/common/executor/ExecutorFactory.java
@@ -0,0 +1,31 @@
+/*
+ * 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.web.engine.flink.common.executor;
+
+/** Factory to create {@link Executor}. */
+public interface ExecutorFactory {
+
+ /**
+ * Creates a new {@link Executor}.
+ *
+ * @return A new instance of {@link Executor}
+ * @throws Exception if executor creation fails
+ */
+ Executor createExecutor() throws Exception;
+}
diff --git a/paimon-web-engine/pom.xml
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/pom.xml
similarity index 72%
copy from paimon-web-engine/pom.xml
copy to
paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/pom.xml
index 81a45f5..3fc87cf 100644
--- a/paimon-web-engine/pom.xml
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/pom.xml
@@ -24,16 +24,18 @@ under the License.
<parent>
<groupId>org.apache.paimon</groupId>
- <artifactId>paimon-webui</artifactId>
+ <artifactId>paimon-web-engine-flink</artifactId>
<version>0.1-SNAPSHOT</version>
</parent>
- <artifactId>paimon-web-engine</artifactId>
- <name>Paimon : Web : Engine</name>
+ <artifactId>paimon-web-engine-flink-sql-gateway</artifactId>
+ <name>Paimon : Web : Engine : Flink : SQL : Gateway</name>
- <packaging>pom</packaging>
-
- <modules>
- <module>paimon-web-engine-flink</module>
- </modules>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-web-engine-flink-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
</project>
\ No newline at end of file
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/client/SqlGatewayClient.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/client/SqlGatewayClient.java
new file mode 100644
index 0000000..f84c699
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/client/SqlGatewayClient.java
@@ -0,0 +1,245 @@
+/*
+ * 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.web.engine.flink.sql.gateway.client;
+
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+import
org.apache.paimon.web.engine.flink.sql.gateway.utils.SqlGateWayRestClient;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import
org.apache.flink.table.gateway.rest.header.operation.CancelOperationHeaders;
+import
org.apache.flink.table.gateway.rest.header.operation.CloseOperationHeaders;
+import
org.apache.flink.table.gateway.rest.header.operation.GetOperationStatusHeaders;
+import org.apache.flink.table.gateway.rest.header.session.CloseSessionHeaders;
+import
org.apache.flink.table.gateway.rest.header.session.ConfigureSessionHeaders;
+import
org.apache.flink.table.gateway.rest.header.session.GetSessionConfigHeaders;
+import org.apache.flink.table.gateway.rest.header.session.OpenSessionHeaders;
+import
org.apache.flink.table.gateway.rest.header.session.TriggerSessionHeartbeatHeaders;
+import
org.apache.flink.table.gateway.rest.header.statement.CompleteStatementHeaders;
+import
org.apache.flink.table.gateway.rest.header.statement.ExecuteStatementHeaders;
+import
org.apache.flink.table.gateway.rest.header.statement.FetchResultsHeaders;
+import
org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import
org.apache.flink.table.gateway.rest.message.session.ConfigureSessionRequestBody;
+import
org.apache.flink.table.gateway.rest.message.session.GetSessionConfigResponseBody;
+import
org.apache.flink.table.gateway.rest.message.session.OpenSessionRequestBody;
+import
org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import
org.apache.flink.table.gateway.rest.message.statement.CompleteStatementRequestBody;
+import
org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import
org.apache.flink.table.gateway.rest.message.statement.FetchResultsMessageParameters;
+import
org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.util.RowFormat;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * The client of flink sql gateway provides some operations of flink sql
gateway. such as creating
+ * session, execute statement, fetch result, etc.
+ */
+public class SqlGatewayClient {
+
+ private static final String DEFAULT_SESSION_NAME_PREFIX =
"FLINK_SQL_GATEWAY_SESSION";
+ private static final int REQUEST_WAITE_TIME = 1000;
+ private static final int ACTIVE_STATUS = 1;
+
+ private final SqlGateWayRestClient restClient;
+ private final String sqlGatewayHost;
+ private final int sqlGatewayPort;
+
+ public SqlGatewayClient(String sqlGatewayHost, int sqlGatewayPort) throws
Exception {
+ this.sqlGatewayHost = sqlGatewayHost;
+ this.sqlGatewayPort = sqlGatewayPort;
+ this.restClient = new SqlGateWayRestClient(sqlGatewayHost,
sqlGatewayPort);
+ }
+
+ public SessionEntity openSession(String sessionName) throws Exception {
+ String name =
+ StringUtils.isBlank(sessionName)
+ ? DEFAULT_SESSION_NAME_PREFIX + "_" + UUID.randomUUID()
+ : sessionName;
+
+ String sessionId =
+ restClient
+ .sendRequest(
+ OpenSessionHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ new OpenSessionRequestBody(name, new
HashMap<>()))
+ .get()
+ .getSessionHandle();
+
+ return SessionEntity.builder()
+ .sessionId(sessionId)
+ .sessionName(sessionName)
+ .host(sqlGatewayHost)
+ .port(sqlGatewayPort)
+ .properties(getSessionConfig(sessionId))
+ .status(ACTIVE_STATUS)
+ .build();
+ }
+
+ public Map<String, String> getSessionConfig(String sessionId) throws
Exception {
+ GetSessionConfigResponseBody getSessionConfigResponseBody =
+ restClient
+ .sendRequest(
+ GetSessionConfigHeaders.getInstance(),
+ new SessionMessageParameters(
+
buildSessionHandleBySessionId(sessionId)),
+ EmptyRequestBody.getInstance())
+ .get();
+ return getSessionConfigResponseBody.getProperties();
+ }
+
+ public void configureSession(String sessionId, String statement, Long
timeout)
+ throws Exception {
+ ConfigureSessionRequestBody configureSessionRequestBody =
+ new ConfigureSessionRequestBody(statement, timeout);
+ restClient
+ .sendRequest(
+ ConfigureSessionHeaders.getInstance(),
+ new
SessionMessageParameters(buildSessionHandleBySessionId(sessionId)),
+ configureSessionRequestBody)
+ .get();
+ }
+
+ public String closeSession(String sessionId) throws Exception {
+ return restClient
+ .sendRequest(
+ CloseSessionHeaders.getInstance(),
+ new
SessionMessageParameters(buildSessionHandleBySessionId(sessionId)),
+ EmptyRequestBody.getInstance())
+ .get()
+ .getStatus();
+ }
+
+ public void triggerSessionHeartbeat(String sessionId) throws Exception {
+ restClient
+ .sendRequest(
+ TriggerSessionHeartbeatHeaders.getInstance(),
+ new
SessionMessageParameters(buildSessionHandleBySessionId(sessionId)),
+ EmptyRequestBody.getInstance())
+ .get();
+ }
+
+ public String executeStatement(String sessionId, String statement,
@Nullable Long timeout)
+ throws Exception {
+ return restClient
+ .sendRequest(
+ ExecuteStatementHeaders.getInstance(),
+ new
SessionMessageParameters(buildSessionHandleBySessionId(sessionId)),
+ new ExecuteStatementRequestBody(statement, timeout,
new HashMap<>()))
+ .get()
+ .getOperationHandle();
+ }
+
+ public List<String> completeStatementHints(String sessionId, String
statement)
+ throws Exception {
+ return restClient
+ .sendRequest(
+ CompleteStatementHeaders.getInstance(),
+ new
SessionMessageParameters(buildSessionHandleBySessionId(sessionId)),
+ new CompleteStatementRequestBody(statement,
statement.length()))
+ .get()
+ .getCandidates();
+ }
+
+ public FetchResultsResponseBody fetchResults(String sessionId, String
operationId, long token)
+ throws Exception {
+ FetchResultsResponseBody fetchResultsResponseBody =
+ restClient
+ .sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
+
buildSessionHandleBySessionId(sessionId),
+
buildOperationHandleByOperationId(operationId),
+ token,
+ RowFormat.JSON),
+ EmptyRequestBody.getInstance())
+ .get();
+ ResultSet.ResultType resultType =
fetchResultsResponseBody.getResultType();
+ while (resultType == ResultSet.ResultType.NOT_READY) {
+ fetchResultsResponseBody =
+ restClient
+ .sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
+
buildSessionHandleBySessionId(sessionId),
+
buildOperationHandleByOperationId(operationId),
+ token,
+ RowFormat.JSON),
+ EmptyRequestBody.getInstance())
+ .get();
+ resultType = fetchResultsResponseBody.getResultType();
+ TimeUnit.MILLISECONDS.sleep(REQUEST_WAITE_TIME);
+ }
+ return fetchResultsResponseBody;
+ }
+
+ public String getOperationStatus(String sessionId, String operationId)
throws Exception {
+ return restClient
+ .sendRequest(
+ GetOperationStatusHeaders.getInstance(),
+ new OperationMessageParameters(
+ buildSessionHandleBySessionId(sessionId),
+
buildOperationHandleByOperationId(operationId)),
+ EmptyRequestBody.getInstance())
+ .get()
+ .getStatus();
+ }
+
+ public String cancelOperation(String sessionId, String operationId) throws
Exception {
+ return restClient
+ .sendRequest(
+ CancelOperationHeaders.getInstance(),
+ new OperationMessageParameters(
+ buildSessionHandleBySessionId(sessionId),
+
buildOperationHandleByOperationId(operationId)),
+ EmptyRequestBody.getInstance())
+ .get()
+ .getStatus();
+ }
+
+ public String closeOperation(String sessionId, String operationId) throws
Exception {
+ return restClient
+ .sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(
+ buildSessionHandleBySessionId(sessionId),
+
buildOperationHandleByOperationId(operationId)),
+ EmptyRequestBody.getInstance())
+ .get()
+ .getStatus();
+ }
+
+ private SessionHandle buildSessionHandleBySessionId(String sessionId) {
+ return new SessionHandle(UUID.fromString(sessionId));
+ }
+
+ private OperationHandle buildOperationHandleByOperationId(String
operationId) {
+ return new OperationHandle(UUID.fromString(operationId));
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/executor/FlinkSqlGatewayExecutor.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/executor/FlinkSqlGatewayExecutor.java
new file mode 100644
index 0000000..12c5b4d
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/executor/FlinkSqlGatewayExecutor.java
@@ -0,0 +1,163 @@
+/*
+ * 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.web.engine.flink.sql.gateway.executor;
+
+import org.apache.paimon.web.engine.flink.common.executor.Executor;
+import
org.apache.paimon.web.engine.flink.common.operation.FlinkSqlOperationType;
+import org.apache.paimon.web.engine.flink.common.parser.StatementParser;
+import org.apache.paimon.web.engine.flink.common.result.ExecutionResult;
+import org.apache.paimon.web.engine.flink.common.result.FetchResultParams;
+import org.apache.paimon.web.engine.flink.sql.gateway.client.SqlGatewayClient;
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+import org.apache.paimon.web.engine.flink.sql.gateway.utils.CollectResultUtil;
+import
org.apache.paimon.web.engine.flink.sql.gateway.utils.FlinkSqlStatementSetBuilder;
+
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import
org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/** The flink sql gateway implementation of the {@link Executor}. */
+public class FlinkSqlGatewayExecutor implements Executor {
+
+ private static final Long DEFAULT_FETCH_TOKEN = 0L;
+ private static final String STOP_JOB_BASE_SQL = "STOP JOB '%s'";
+ private static final String WITH_SAVEPOINT = " WITH SAVEPOINT";
+
+ private final SqlGatewayClient client;
+ private final SessionEntity session;
+
+ public FlinkSqlGatewayExecutor(SessionEntity session) throws Exception {
+ this.session = session;
+ this.client = new SqlGatewayClient(session.getHost(),
session.getPort());
+ }
+
+ @Override
+ public ExecutionResult executeSql(String multiStatement) throws Exception {
+ String[] statements = StatementParser.parse(multiStatement);
+ List<String> insertStatements = new ArrayList<>();
+ ExecutionResult executionResult = null;
+
+ for (String statement : statements) {
+ FlinkSqlOperationType operationType =
FlinkSqlOperationType.getOperationType(statement);
+
+ if (operationType == null) {
+ String operationTypeString =
extractSqlOperationType(statement);
+ throw new UnsupportedOperationException(
+ "Unsupported operation type: " + operationTypeString);
+ }
+
+ switch (operationType.getCategory()) {
+ case DQL:
+ if (!insertStatements.isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Cannot execute DQL statement with pending
INSERT statements.");
+ }
+ executionResult = executeDqlStatement(statement,
operationType);
+ break;
+ case DML:
+ if
(operationType.getType().equals(FlinkSqlOperationType.INSERT.getType())) {
+ insertStatements.add(statement);
+ } else {
+ executionResult = executeDmlStatement(statement);
+ }
+ break;
+ default:
+ client.executeStatement(session.getSessionId(), statement,
null);
+ break;
+ }
+
+ if (executionResult != null) {
+ return executionResult;
+ }
+ }
+
+ if (!insertStatements.isEmpty()) {
+ String combinedStatement =
+
FlinkSqlStatementSetBuilder.buildStatementSet(insertStatements);
+ executionResult = executeDmlStatement(combinedStatement);
+ }
+
+ return executionResult;
+ }
+
+ private ExecutionResult executeDqlStatement(
+ String statement, FlinkSqlOperationType operationType) throws
Exception {
+ String operationId = client.executeStatement(session.getSessionId(),
statement, null);
+ FetchResultsResponseBody results =
+ client.fetchResults(session.getSessionId(), operationId,
DEFAULT_FETCH_TOKEN);
+ ExecutionResult.Builder builder =
+ CollectResultUtil.collectSqlGatewayResult(results.getResults())
+ .submitId(operationId);
+ if
(operationType.getType().equals(FlinkSqlOperationType.SELECT.getType())) {
+
builder.jobId(getJobIdFromResults(results)).shouldFetchResult(true);
+ }
+ return builder.build();
+ }
+
+ private ExecutionResult executeDmlStatement(String statement) throws
Exception {
+ String operationId = client.executeStatement(session.getSessionId(),
statement, null);
+ FetchResultsResponseBody results =
+ client.fetchResults(session.getSessionId(), operationId,
DEFAULT_FETCH_TOKEN);
+ return new ExecutionResult.Builder()
+ .submitId(operationId)
+ .jobId(getJobIdFromResults(results))
+ .build();
+ }
+
+ private String getJobIdFromResults(FetchResultsResponseBody results) {
+ return Objects.requireNonNull(results.getJobID(), "Job ID not found in
results").toString();
+ }
+
+ @Override
+ public ExecutionResult fetchResults(FetchResultParams params) throws
Exception {
+ FetchResultsResponseBody fetchResultsResponseBody =
+ client.fetchResults(params.getSessionId(),
params.getSubmitId(), params.getToken());
+ ResultSet.ResultType resultType =
fetchResultsResponseBody.getResultType();
+ if (resultType == ResultSet.ResultType.EOS) {
+ return ExecutionResult.builder().shouldFetchResult(false).build();
+ }
+ ExecutionResult.Builder builder =
+
CollectResultUtil.collectSqlGatewayResult(fetchResultsResponseBody.getResults());
+ builder.submitId(params.getSubmitId());
+ return builder.build();
+ }
+
+ @Override
+ public void stop(String jobId, boolean withSavepoint) throws Exception {
+ StringBuilder sqlBuilder = new
StringBuilder(String.format(STOP_JOB_BASE_SQL, jobId));
+ if (withSavepoint) {
+ sqlBuilder.append(WITH_SAVEPOINT);
+ }
+ client.executeStatement(session.getSessionId(), sqlBuilder.toString(),
null);
+ }
+
+ private String extractSqlOperationType(String sql) {
+ Pattern pattern = Pattern.compile("^(\\w+)");
+ Matcher matcher = pattern.matcher(sql);
+ if (matcher.find()) {
+ return matcher.group(1);
+ }
+ return "UNKNOWN";
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/executor/FlinkSqlGatewayExecutorFactory.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/executor/FlinkSqlGatewayExecutorFactory.java
new file mode 100644
index 0000000..be05dde
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/executor/FlinkSqlGatewayExecutorFactory.java
@@ -0,0 +1,38 @@
+/*
+ * 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.web.engine.flink.sql.gateway.executor;
+
+import org.apache.paimon.web.engine.flink.common.executor.Executor;
+import org.apache.paimon.web.engine.flink.common.executor.ExecutorFactory;
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+
+/** Factory to create {@link FlinkSqlGatewayExecutor}. */
+public class FlinkSqlGatewayExecutorFactory implements ExecutorFactory {
+
+ private final SessionEntity sessionEntity;
+
+ public FlinkSqlGatewayExecutorFactory(SessionEntity sessionEntity) {
+ this.sessionEntity = sessionEntity;
+ }
+
+ @Override
+ public Executor createExecutor() throws Exception {
+ return new FlinkSqlGatewayExecutor(sessionEntity);
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/model/SessionEntity.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/model/SessionEntity.java
new file mode 100644
index 0000000..c0bf8a0
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/model/SessionEntity.java
@@ -0,0 +1,37 @@
+/*
+ * 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.web.engine.flink.sql.gateway.model;
+
+import lombok.Builder;
+import lombok.Getter;
+
+import java.util.Map;
+
+/** The session entity. */
+@Builder
+@Getter
+public class SessionEntity {
+
+ private final String sessionId;
+ private final String sessionName;
+ private final String host;
+ private final int port;
+ private final Map<String, String> properties;
+ private final int status;
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/CollectResultUtil.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/CollectResultUtil.java
new file mode 100644
index 0000000..8ec6f45
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/CollectResultUtil.java
@@ -0,0 +1,63 @@
+/*
+ * 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.web.engine.flink.sql.gateway.utils;
+
+import org.apache.paimon.web.engine.flink.common.result.ExecutionResult;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.gateway.rest.serde.ResultInfo;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/** Collect result util. */
+public class CollectResultUtil {
+
+ public static ExecutionResult.Builder collectSqlGatewayResult(ResultInfo
resultInfo) {
+ List<RowData> data = resultInfo.getData();
+ List<Map<String, Object>> results =
+ rowDatasToList(resultInfo.getResultSchema().getColumnNames(),
data);
+ return ExecutionResult.builder().data(results);
+ }
+
+ private static List<Map<String, Object>> rowDatasToList(
+ List<String> columns, List<RowData> rowDataList) {
+ List<Map<String, Object>> rows = new ArrayList<>();
+ for (RowData rowData : rowDataList) {
+ Map<String, Object> map = new LinkedHashMap<>();
+ for (int i = 0; i < rowData.getArity(); ++i) {
+ if (rowData instanceof GenericRowData) {
+ GenericRowData data = (GenericRowData) rowData;
+ Object field = data.getField(i);
+ if (Objects.nonNull(field)) {
+ map.put(columns.get(i), field.toString());
+ }
+ } else {
+ throw new IllegalArgumentException("RowData is not
GenericData.");
+ }
+ }
+ rows.add(map);
+ }
+ return rows;
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/FlinkSqlStatementSetBuilder.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/FlinkSqlStatementSetBuilder.java
new file mode 100644
index 0000000..637de23
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/FlinkSqlStatementSetBuilder.java
@@ -0,0 +1,61 @@
+/*
+ * 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.web.engine.flink.sql.gateway.utils;
+
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.List;
+
+/**
+ * Utility class for building Flink SQL STATEMENT SET strings for batch
execution.
+ *
+ * <p>This class provides a static method to construct a STATEMENT SET from a
list of individual
+ * INSERT statements. If there is only one INSERT statement, it returns the
statement as is. For
+ * multiple statements, it wraps them in a Flink SQL EXECUTE STATEMENT SET
block.
+ */
+public class FlinkSqlStatementSetBuilder {
+
+ private static final String EXECUTE_STATEMENT_SET = "EXECUTE STATEMENT
SET\n";
+ private static final String BEGIN_STATEMENT = "BEGIN\n";
+ private static final String END_STATEMENT = "END;\n";
+ private static final String STATEMENT_DELIMITER = ";\n";
+
+ public static String buildStatementSet(List<String> insertStatements) {
+ StringBuilder statementSetBuilder = new StringBuilder();
+
+ if (CollectionUtils.isNotEmpty(insertStatements)) {
+ if (insertStatements.size() > 1) {
+ statementSetBuilder.append(EXECUTE_STATEMENT_SET);
+ statementSetBuilder.append(BEGIN_STATEMENT);
+
+ for (String insertStatement : insertStatements) {
+ statementSetBuilder.append(insertStatement);
+ statementSetBuilder.append(STATEMENT_DELIMITER);
+ }
+
+ statementSetBuilder.append(END_STATEMENT);
+ } else {
+ statementSetBuilder.append(insertStatements.get(0));
+ statementSetBuilder.append(STATEMENT_DELIMITER);
+ }
+ }
+
+ return statementSetBuilder.toString();
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/SqlGateWayRestClient.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/SqlGateWayRestClient.java
new file mode 100644
index 0000000..4679b59
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/main/java/org/apache/paimon/web/engine/flink/sql/gateway/utils/SqlGateWayRestClient.java
@@ -0,0 +1,147 @@
+/*
+ * 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.web.engine.flink.sql.gateway.utils;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.FileUpload;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.versioning.RestAPIVersion;
+import org.apache.flink.util.ConfigurationException;
+import org.apache.flink.util.concurrent.Executors;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+/** SQL gateway rest client. */
+public class SqlGateWayRestClient extends RestClient {
+ private final String sqlGatewayHost;
+ private final int sqlGatewayPort;
+
+ public SqlGateWayRestClient(String sqlGatewayHost, int sqlGatewayPort)
+ throws ConfigurationException {
+ super(new Configuration(), Executors.newDirectExecutorService());
+ this.sqlGatewayHost = sqlGatewayHost;
+ this.sqlGatewayPort = sqlGatewayPort;
+ }
+
+ public <
+ M extends MessageHeaders<R, P, U>,
+ U extends MessageParameters,
+ R extends RequestBody,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(
+ M messageHeaders,
+ U messageParameters,
+ R request,
+ Collection<FileUpload> fileUploads,
+ RestAPIVersion<? extends RestAPIVersion<?>> apiVersion) {
+ return throwExceptionHandler(
+ () -> {
+ try {
+ return super.sendRequest(
+ sqlGatewayHost,
+ sqlGatewayPort,
+ messageHeaders,
+ messageParameters,
+ request,
+ fileUploads,
+ apiVersion);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ public <
+ M extends MessageHeaders<R, P, U>,
+ U extends MessageParameters,
+ R extends RequestBody,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(M messageHeaders, U
messageParameters, R request) {
+ return throwExceptionHandler(
+ () -> {
+ try {
+ return super.sendRequest(
+ sqlGatewayHost,
+ sqlGatewayPort,
+ messageHeaders,
+ messageParameters,
+ request);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ public <
+ M extends MessageHeaders<R, P, U>,
+ U extends MessageParameters,
+ R extends RequestBody,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(
+ M messageHeaders,
+ U messageParameters,
+ R request,
+ Collection<FileUpload> fileUploads) {
+ return throwExceptionHandler(
+ () -> {
+ try {
+ return super.sendRequest(
+ sqlGatewayHost,
+ sqlGatewayPort,
+ messageHeaders,
+ messageParameters,
+ request,
+ fileUploads);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ public <
+ M extends MessageHeaders<EmptyRequestBody, P,
EmptyMessageParameters>,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(M messageHeaders) {
+ return throwExceptionHandler(
+ () -> {
+ try {
+ return super.sendRequest(sqlGatewayHost,
sqlGatewayPort, messageHeaders);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ private <T> T throwExceptionHandler(Supplier<T> func0) {
+ try {
+ return func0.get();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/TestBase.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/TestBase.java
new file mode 100644
index 0000000..32e3a3f
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/TestBase.java
@@ -0,0 +1,201 @@
+/*
+ * 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.web.engine.flink.sql.gataway;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import
org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils;
+import org.apache.flink.table.gateway.rest.SqlGatewayRestEndpoint;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestOptions;
+import org.apache.flink.table.gateway.service.SqlGatewayServiceImpl;
+import org.apache.flink.table.gateway.service.context.DefaultContext;
+import org.apache.flink.table.gateway.service.session.SessionManager;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.rules.TemporaryFolder;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import static
org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
+import static
org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getEndpointConfig;
+import static
org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getSqlGatewayOptionPrefix;
+import static
org.apache.flink.table.gateway.rest.SqlGatewayRestEndpointFactory.IDENTIFIER;
+import static
org.apache.flink.table.gateway.rest.SqlGatewayRestEndpointFactory.rebuildRestEndpointOptions;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** The base class for sql gateway test. */
+public class TestBase {
+
+ @RegisterExtension
+ @Order(1)
+ private static final MiniClusterExtension MINI_CLUSTER = new
MiniClusterExtension();
+
+ @RegisterExtension
+ @Order(2)
+ protected static final SqlGatewayServiceExtension
SQL_GATEWAY_SERVICE_EXTENSION =
+ new
SqlGatewayServiceExtension(MINI_CLUSTER::getClientConfiguration);
+
+ @Nullable protected static String targetAddress = null;
+ @Nullable private static SqlGatewayRestEndpoint sqlGatewayRestEndpoint =
null;
+
+ protected static int port = 0;
+
+ @BeforeAll
+ static void start() throws Exception {
+ final String address =
InetAddress.getLoopbackAddress().getHostAddress();
+ Configuration config = getBaseConfig(getFlinkConfig(address, address,
"0"));
+ sqlGatewayRestEndpoint =
+ new SqlGatewayRestEndpoint(config,
SQL_GATEWAY_SERVICE_EXTENSION.getService());
+ sqlGatewayRestEndpoint.start();
+ InetSocketAddress serverAddress =
checkNotNull(sqlGatewayRestEndpoint.getServerAddress());
+ targetAddress = serverAddress.getHostName();
+ port = serverAddress.getPort();
+ }
+
+ @AfterAll
+ static void stop() throws Exception {
+ checkNotNull(sqlGatewayRestEndpoint);
+ sqlGatewayRestEndpoint.close();
+ }
+
+ private static Configuration getBaseConfig(Configuration flinkConf) {
+ SqlGatewayEndpointFactoryUtils.DefaultEndpointFactoryContext context =
+ new
SqlGatewayEndpointFactoryUtils.DefaultEndpointFactoryContext(
+ null, flinkConf, getEndpointConfig(flinkConf,
IDENTIFIER));
+
+ return rebuildRestEndpointOptions(context.getEndpointOptions());
+ }
+
+ private static Configuration getFlinkConfig(
+ String address, String bindAddress, String portRange) {
+ final Configuration config = new Configuration();
+ if (address != null) {
+ config.setString(
+
getSqlGatewayRestOptionFullName(SqlGatewayRestOptions.ADDRESS.key()), address);
+ }
+ if (bindAddress != null) {
+ config.setString(
+
getSqlGatewayRestOptionFullName(SqlGatewayRestOptions.BIND_ADDRESS.key()),
+ bindAddress);
+ }
+ if (portRange != null) {
+ config.setString(
+
getSqlGatewayRestOptionFullName(SqlGatewayRestOptions.PORT.key()), portRange);
+ }
+ return config;
+ }
+
+ private static String getSqlGatewayRestOptionFullName(String key) {
+ return getSqlGatewayOptionPrefix(IDENTIFIER) + key;
+ }
+
+ /** A simple {@link Extension} to be used by tests that require a {@link
SqlGatewayService}. */
+ static class SqlGatewayServiceExtension implements BeforeAllCallback,
AfterAllCallback {
+
+ private SqlGatewayService service;
+ private SessionManager sessionManager;
+ private TemporaryFolder temporaryFolder;
+ private final Supplier<Configuration> configSupplier;
+ private final Function<DefaultContext, SessionManager>
sessionManagerCreator;
+
+ public SqlGatewayServiceExtension(Supplier<Configuration>
configSupplier) {
+ this(configSupplier, SessionManager::create);
+ }
+
+ public SqlGatewayServiceExtension(
+ Supplier<Configuration> configSupplier,
+ Function<DefaultContext, SessionManager>
sessionManagerCreator) {
+ this.configSupplier = configSupplier;
+ this.sessionManagerCreator = sessionManagerCreator;
+ }
+
+ @Override
+ public void beforeAll(ExtensionContext context) throws Exception {
+ final Map<String, String> originalEnv = System.getenv();
+ try {
+ // prepare conf dir
+ temporaryFolder = new TemporaryFolder();
+ temporaryFolder.create();
+ File confFolder = temporaryFolder.newFolder("conf");
+ File confYaml = new File(confFolder, "flink-conf.yaml");
+ if (!confYaml.createNewFile()) {
+ throw new IOException("Can't create testing
flink-conf.yaml file.");
+ }
+
+ FileUtils.write(
+ confYaml,
+ getFlinkConfContent(configSupplier.get().toMap()),
+ StandardCharsets.UTF_8);
+
+ // adjust the test environment for the purposes of this test
+ Map<String, String> map = new HashMap<>(System.getenv());
+ map.put(ENV_FLINK_CONF_DIR, confFolder.getAbsolutePath());
+ CommonTestUtils.setEnv(map);
+
+ sessionManager =
+ sessionManagerCreator.apply(
+ DefaultContext.load(
+ new Configuration(),
Collections.emptyList(), true, false));
+ } finally {
+ CommonTestUtils.setEnv(originalEnv);
+ }
+
+ service = new SqlGatewayServiceImpl(sessionManager);
+ sessionManager.start();
+ }
+
+ @Override
+ public void afterAll(ExtensionContext context) throws Exception {
+ if (sessionManager != null) {
+ sessionManager.stop();
+ }
+ temporaryFolder.delete();
+ }
+
+ public SqlGatewayService getService() {
+ return service;
+ }
+
+ private String getFlinkConfContent(Map<String, String> flinkConf) {
+ StringBuilder sb = new StringBuilder();
+ flinkConf.forEach((k, v) -> sb.append(k).append(":
").append(v).append("\n"));
+ return sb.toString();
+ }
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/client/SqlGatewayClientTest.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/client/SqlGatewayClientTest.java
new file mode 100644
index 0000000..1b20e31
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/client/SqlGatewayClientTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.web.engine.flink.sql.gataway.client;
+
+import org.apache.paimon.web.engine.flink.sql.gataway.TestBase;
+import org.apache.paimon.web.engine.flink.sql.gateway.client.SqlGatewayClient;
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+
+import org.apache.commons.collections.MapUtils;
+import
org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link SqlGatewayClient}. */
+public class SqlGatewayClientTest extends TestBase {
+
+ SqlGatewayClient client;
+ SessionEntity session;
+
+ private static final String SESSION_NAME = "test_session";
+
+ @BeforeEach
+ void before() throws Exception {
+ client = new SqlGatewayClient(targetAddress, port);
+ session = client.openSession(SESSION_NAME);
+ }
+
+ @Test
+ public void testGetSessionConfig() throws Exception {
+ Map<String, String> sessionConfig =
client.getSessionConfig(session.getSessionId());
+ assertTrue(MapUtils.isNotEmpty(sessionConfig));
+ }
+
+ @Test
+ public void testCloseSession() throws Exception {
+ String status = client.closeSession(session.getSessionId());
+ assertEquals("CLOSED", status);
+ }
+
+ @Test
+ public void testExecuteStatement() throws Exception {
+ String operationHandle =
client.executeStatement(session.getSessionId(), "SELECT 1", null);
+ assertNotNull(operationHandle);
+ }
+
+ @Test
+ public void testCompleteStatementHints() throws Exception {
+ List<String> list =
client.completeStatementHints(session.getSessionId(), "CREATE TA");
+ assertFalse(list.isEmpty());
+ }
+
+ @Test
+ public void testFetchResults() throws Exception {
+ String operationHandle =
client.executeStatement(session.getSessionId(), "SELECT 1", null);
+ FetchResultsResponseBody fetchResultsResponseBody =
+ client.fetchResults(session.getSessionId(), operationHandle,
0);
+ assertNotNull(fetchResultsResponseBody);
+ assertEquals("PAYLOAD",
fetchResultsResponseBody.getResultType().name());
+ FetchResultsResponseBody fetchResultsResponseBodyNext =
+ client.fetchResults(session.getSessionId(), operationHandle,
1);
+ assertNotNull(fetchResultsResponseBodyNext);
+ assertEquals("EOS",
fetchResultsResponseBodyNext.getResultType().name());
+ }
+
+ @Test
+ public void testGetOperationStatus() throws Exception {
+ String operationHandle =
client.executeStatement(session.getSessionId(), "SELECT 1", null);
+ String operationStatus =
client.getOperationStatus(session.getSessionId(), operationHandle);
+ assertNotNull(operationStatus);
+ }
+
+ @Test
+ public void testCancelOperation() throws Exception {
+ String operationHandle =
client.executeStatement(session.getSessionId(), "SELECT 1", null);
+ String status = client.cancelOperation(session.getSessionId(),
operationHandle);
+ assertEquals("CANCELED", status);
+ }
+
+ @Test
+ public void testCloseOperation() throws Exception {
+ String operationHandle =
client.executeStatement(session.getSessionId(), "SELECT 1", null);
+ String status = client.closeOperation(session.getSessionId(),
operationHandle);
+ assertEquals("CLOSED", status);
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/executor/FlinkSqlGatewayExecutorTest.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/executor/FlinkSqlGatewayExecutorTest.java
new file mode 100644
index 0000000..42620af
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/executor/FlinkSqlGatewayExecutorTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.web.engine.flink.sql.gataway.executor;
+
+import org.apache.paimon.web.engine.flink.common.result.ExecutionResult;
+import org.apache.paimon.web.engine.flink.common.result.FetchResultParams;
+import org.apache.paimon.web.engine.flink.sql.gataway.TestBase;
+import org.apache.paimon.web.engine.flink.sql.gateway.client.SqlGatewayClient;
+import
org.apache.paimon.web.engine.flink.sql.gateway.executor.FlinkSqlGatewayExecutor;
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link FlinkSqlGatewayExecutor}. */
+public class FlinkSqlGatewayExecutorTest extends TestBase {
+
+ SqlGatewayClient client;
+ SessionEntity session;
+ FlinkSqlGatewayExecutor executor;
+
+ private static final String SESSION_NAME = "test_session";
+
+ @BeforeEach
+ void before() throws Exception {
+ client = new SqlGatewayClient(targetAddress, port);
+ session = client.openSession(SESSION_NAME);
+ executor = new FlinkSqlGatewayExecutor(session);
+ }
+
+ @Test
+ public void testExecuteSql() throws Exception {
+ ExecutionResult executionResult =
executor.executeSql(StatementsConstant.statement);
+ assertNotNull(executionResult);
+ assertNotNull(executionResult.getJobId());
+ }
+
+ @Test
+ public void testExecuteStatementSetSql() throws Exception {
+ ExecutionResult executionResult =
executor.executeSql(StatementsConstant.statementSetSql);
+ assertNotNull(executionResult);
+ assertNotNull(executionResult.getJobId());
+ }
+
+ @Test
+ public void testExecutorStatementWithoutResult() throws Exception {
+ ExecutionResult executionResult =
executor.executeSql(StatementsConstant.createStatement);
+ assertNull(executionResult);
+ }
+
+ @Test
+ public void testExecuteDQLStatementWithPendingInsertStatements() {
+ Exception exception =
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> {
+ executor.executeSql(
+
StatementsConstant.selectStatementWithPendingInsertStatements);
+ });
+ String expectedMessage = "Cannot execute DQL statement with pending
INSERT statements.";
+ String actualMessage = exception.getMessage();
+ assertTrue(actualMessage.contains(expectedMessage));
+ }
+
+ @Test
+ public void testExecuteBadSqlStatement() {
+ Exception exception =
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> {
+
executor.executeSql(StatementsConstant.badStatement);
+ });
+ String expectedMessage = "Unsupported operation type: CREAT";
+ String actualMessage = exception.getMessage();
+ assertTrue(actualMessage.contains(expectedMessage));
+ }
+
+ @Test
+ public void testFetchResults() throws Exception {
+ ExecutionResult executionResult =
executor.executeSql(StatementsConstant.selectStatement);
+ assertNotNull(executionResult);
+ assertNotNull(executionResult.getJobId());
+ assertNotNull(executionResult.getSubmitId());
+ assertTrue(executionResult.shouldFetchResult());
+ FetchResultParams params =
+ FetchResultParams.builder()
+ .sessionId(session.getSessionId())
+ .submitId(executionResult.getSubmitId())
+ .token(1L)
+ .build();
+ ExecutionResult fetchResult = executor.fetchResults(params);
+ assertFalse(fetchResult.getData().isEmpty());
+ }
+}
diff --git
a/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/executor/StatementsConstant.java
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/executor/StatementsConstant.java
new file mode 100644
index 0000000..c08f05a
--- /dev/null
+++
b/paimon-web-engine/paimon-web-engine-flink/paimon-web-engine-flink-sql-gateway/src/test/java/org/apache/paimon/web/engine/flink/sql/gataway/executor/StatementsConstant.java
@@ -0,0 +1,196 @@
+/*
+ * 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.web.engine.flink.sql.gataway.executor;
+
+/** Statements constant. */
+public class StatementsConstant {
+
+ public static String statement =
+ "CREATE TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n"
+ + "CREATE TABLE IF NOT EXISTS sink_table(\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` TIMESTAMP(3),\n"
+ + " `one_minute_sum` BIGINT\n"
+ + ") WITH('connector' = 'print');\n"
+ + "\n"
+ + "INSERT INTO\n"
+ + " sink_table\n"
+ + "SELECT\n"
+ + " product,\n"
+ + " amount,\n"
+ + " order_time,\n"
+ + " 0 as one_minute_sum\n"
+ + "FROM\n"
+ + " t_order;";
+
+ public static String selectStatement =
+ "CREATE TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n"
+ + "SELECT * FROM t_order;";
+
+ public static String selectStatementWithPendingInsertStatements =
+ "CREATE TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n"
+ + "CREATE TABLE IF NOT EXISTS sink_table(\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` TIMESTAMP(3),\n"
+ + " `one_minute_sum` BIGINT\n"
+ + ") WITH('connector' = 'print');\n"
+ + "\n"
+ + "INSERT INTO\n"
+ + " sink_table\n"
+ + "SELECT\n"
+ + " product,\n"
+ + " amount,\n"
+ + " order_time,\n"
+ + " 0 as one_minute_sum\n"
+ + "FROM\n"
+ + " t_order;\n"
+ + "SELECT * FROM t_order LIMIT 10;";
+
+ public static String createStatement =
+ "CREATE TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n";
+
+ public static String badStatement =
+ "CREAT TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n";
+
+ public static String statementSetSql =
+ "CREATE TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n"
+ + "CREATE TABLE IF NOT EXISTS sink_table_01(\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` TIMESTAMP(3),\n"
+ + " `one_minute_sum` BIGINT\n"
+ + ") WITH('connector' = 'print');\n"
+ + "CREATE TABLE IF NOT EXISTS sink_table_02(\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` TIMESTAMP(3),\n"
+ + " `one_minute_sum` BIGINT\n"
+ + ") WITH('connector' = 'print');\n"
+ + "\n"
+ + "INSERT INTO\n"
+ + " sink_table_01\n"
+ + "SELECT\n"
+ + " product,\n"
+ + " amount,\n"
+ + " order_time,\n"
+ + " 0 as one_minute_sum\n"
+ + "FROM\n"
+ + " t_order;\n"
+ + "INSERT INTO\n"
+ + " sink_table_02\n"
+ + "SELECT\n"
+ + " product,\n"
+ + " amount,\n"
+ + " order_time,\n"
+ + " 0 as one_minute_sum\n"
+ + "FROM\n"
+ + " t_order;";
+}
diff --git a/paimon-web-engine/paimon-web-engine-flink/pom.xml
b/paimon-web-engine/paimon-web-engine-flink/pom.xml
index fa44e1c..9365cff 100644
--- a/paimon-web-engine/paimon-web-engine-flink/pom.xml
+++ b/paimon-web-engine/paimon-web-engine-flink/pom.xml
@@ -35,6 +35,7 @@ under the License.
<modules>
<module>paimon-web-engine-flink-common</module>
+ <module>paimon-web-engine-flink-sql-gateway</module>
</modules>
<properties>
diff --git a/paimon-web-engine/pom.xml b/paimon-web-engine/pom.xml
index 81a45f5..1070298 100644
--- a/paimon-web-engine/pom.xml
+++ b/paimon-web-engine/pom.xml
@@ -36,4 +36,12 @@ under the License.
<modules>
<module>paimon-web-engine-flink</module>
</modules>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-web-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
</project>
\ No newline at end of file
